Frameworks

Every major agent framework now has a story for multi-agent systems. Most of them are incompatible with each other. An agent built in AutoGen cannot natively receive a task from a deepagents orchestrator. An OpenAI Agents SDK cannot talk to a LangGraph subgraph. A CrewAI crew cannot delegate to a Pydantic AI team without custom glue code. This is the inter-agent protocol problem: We have multi-agent frameworks, but no agreed-upon protocol for agents to communicate across frameworks. This post breaks down the four main approaches, compares them, and examines the need for standardization. Why Is Inter-Agent Communication Hard? Single-agent systems have a clean interface: you send a prompt, you get a response. Multi-agent systems need to express: Task delegation – what work is being handed off and to whomContext transfer – what is the background the receiving agent needsState propagation – what the delegating agent needs backError and cancellation – what happens when the receiving agent failsStreaming – how partial results flow back during long tasks Most frameworks solve some of these, but with different schemas, different transport assumptions, and no interoperability. The 4 Current Approaches 1. ACP (Agent Communication Protocol): deepagents Installation: Shell pip install deepagents-acp # Spec: https://github.com/langchain-ai/deepagents/tree/main/libs/acp # deepagents-acp version: 0.0.6 (requires agent-client-protocol>=0.8.0) ACP is an open protocol: a published JSON schema for agent-to-agent communication. Langchain's deepagents ships deepagents-acp, a Python client and server implementation, as a separate package, so any framework can implement it. ACP Message Flow A deepagents orchestrator routing a task to an ACP-compatible worker: Python from deepagents import create_deep_agent from deepagents.middleware import AsyncSubAgentMiddleware, AsyncSubAgent # Declare an ACP-compatible worker agent research_worker = AsyncSubAgent( name="research", description="Deep research agent — use for tasks requiring web search and synthesis", url="http://research-agent:8080", # ACP server endpoint ) # The orchestrator routes tasks to workers via the ACP protocol automatically orchestrator = create_deep_agent( model="anthropic:claude-sonnet-4-6", middleware=[ AsyncSubAgentMiddleware(subagents=[research_worker]), ], ) Serving a deepagents agent as an ACP endpoint: Python import asyncio from acp import run_agent as run_acp_agent from deepagents import create_deep_agent from deepagents_acp.server import AgentServerACP agent = create_deep_agent(model="anthropic:claude-haiku-4-5") # Wrap the compiled graph and expose it over ACP acp_agent = AgentServerACP(agent=agent) asyncio.run(run_acp_agent(acp_agent)) What makes ACP different: It's an open schema, not a framework-internal call. Any framework can implement an ACP server or client, which means a CrewAI crew could delegate to a deepagents worker over ACP without any shared code. Current limitation: Adoption is early. As of v0.9.0, the primary implementations are deepagents-native. Support for other frameworks requires each to implement the server interface independently. 2. Handoffs: OpenAI Agents SDK Installation: Shell pip install openai-agents # Docs: https://openai.github.io/openai-agents-python/handoffs/ OpenAI's SDK handles delegation through handoffs: an agent declares which other agents it can hand off to. At runtime, the orchestrator agent decides when to delegate and execution transfers. Python from agents import Agent, Runner, handoff research_agent = Agent( name="ResearchAgent", instructions="You are an expert at web research. Answer research questions thoroughly.", tools=[web_search_tool], ) code_agent = Agent( name="CodeAgent", instructions="You write and review Python code.", tools=[run_code_tool], ) orchestrator = Agent( name="Orchestrator", instructions=( "Route tasks to the right specialist. " "Use ResearchAgent for questions requiring web search. " "Use CodeAgent for programming tasks." ), handoffs=[ handoff(research_agent), handoff(code_agent, tool_name_override="delegate_to_coder"), ], ) result = await Runner.run(orchestrator, "Write a Python script to fetch weather data") You can also add context and filters to handoffs: Python from agents import handoff, RunContextWrapper def on_handoff_to_research(ctx: RunContextWrapper, input_data: str) -> None: print(f"Handing off to research agent with: {input_data}") research_handoff = handoff( research_agent, on_handoff=on_handoff_to_research, input_filter=lambda inp: inp, # transform input before handoff ) Limitation: Handoffs are framework-internal. A handoff target must be an Agent instance from the same SDK. There's no published schema, no HTTP transport, no cross-framework interoperability. If your orchestrator is a LangGraph graph and your specialist is an OpenAI agent, handoffs can't bridge them. 3. Agent Delegation via Tools: Pydantic AI Installation: Shell pip install pydantic-ai # Docs: https://pydantic.dev/docs/ai/guides/multi-agent-applications/ Pydantic AI does not have a dedicated inter-agent messaging primitive. The idiomatic multi-agent pattern is agent delegation via tools: a specialist agent is called from within a tool of the orchestrator agent, and the result is returned to the orchestrator like any other tool return value. Python from pydantic_ai import Agent, RunContext research_agent = Agent( "anthropic:claude-haiku-4-5", system_prompt="You are a research specialist. Answer questions with citations.", ) code_agent = Agent( "anthropic:claude-sonnet-4-6", system_prompt="You are a Python expert. Write clean, tested code.", ) orchestrator = Agent( "anthropic:claude-sonnet-4-6", system_prompt="Route tasks to the right specialist tool.", ) @orchestrator.tool async def research(ctx: RunContext[None], query: str) -> str: """Delegate a research question to the research specialist.""" result = await research_agent.run(query, usage=ctx.usage) return result.output @orchestrator.tool async def write_code(ctx: RunContext[None], task: str) -> str: """Delegate a coding task to the code specialist.""" result = await code_agent.run(task, usage=ctx.usage) return result.output result = await orchestrator.run( "Research quantum computing and write a Python simulation" ) Passing usage=ctx.usage propagates token accounting from the delegate run back to the parent, so result.usage() that the orchestrator covers all sub-agent calls. Limitation: Delegation is framework-internal: all agents must be Pydantic AI Agent instances. There is no pub-sub bus, no HTTP transport, and no cross-framework protocol. For true cross-framework delegation, a pattern like ACP is required. 4. AgentProtocol Backbone: Agno Installation: Shell pip install agno # Docs: https://docs.agno.com/introduction Agno takes the most ambitious approach: an AgentProtocol backbone that acts as a multi-framework adapter. The goal is to let agents from different frameworks (LangGraph, DSPy, Claude's SDK) plug into the same Agno team: Python from agno.agent import Agent from agno.team import Team from agno.models.anthropic import Claude # Native Agno agent research_agent = Agent( name="Researcher", model=Claude(id="claude-haiku-4-5"), tools=[web_search_tool], description="Specializes in web research and synthesis.", ) # Native Agno agent code_agent = Agent( name="Coder", model=Claude(id="claude-sonnet-4-6"), tools=[python_repl_tool], description="Specializes in Python development.", ) team = Team( name="FullStackTeam", mode="coordinate", members=[research_agent, code_agent], model=Claude(id="claude-sonnet-4-6"), db=agno_db_storage, # sessions + memory persistence enable_agentic_state=True, ) team.print_response( "Research quantum computing trends and prototype a simulation in Python" ) Limitation: The multi-framework adapter is still maturing. Most production Agno deployments use native Agno agents. The cross-framework vision is real, but the published adapter surface for LangGraph and DSPy is sparse at the time of writing. Side-by-Side Comparison DimensionACP (deepagents)Handoffs (OpenAI)Agent delegation (Pydantic AI)AgentProtocol (Agno)Published open schema✓✗✗PartialHTTP transport✓✗✗✓Streaming support✓ (SSE)✓ (run-level)✗✓Cross-framework workers✓ (any ACP server)✗✗PartialContext/metadata passing✓✓ (input_filter)✓ (usage propagation)✓Error/cancellation schema✓Partial✗✓Built-in state persistence✗ (LangGraph handles it)✗✗✓ (Agno DB)Production deploymentsEarlyGrowingMatureGrowing The Fragmentation Cost in Practice Consider this real scenario: you have a research pipeline where: The orchestrator is a deepagents agent (LangGraph-backed)The research worker is a CrewAI crew (good at parallel research tasks)The code worker is an OpenAI agent (good at code + sandboxed execution) Today, wiring this up requires: Python # Option 1: Wrap every non-deepagents agent as a plain function tool # Loses: streaming, cancellation, structured error handling @tool def run_crewai_research(query: str) -> str: crew = ResearchCrew() result = crew.kickoff(inputs={"query": query}) return str(result) # no streaming, no structured output # Option 2: Host each agent as an HTTP service and call it manually # Loses: shared context, standard error handling, progress tracking import httpx @tool async def call_openai_agent(task: str) -> str: async with httpx.AsyncClient() as client: response = await client.post( "http://openai-agent-service/run", json={"task": task}, ) return response.json()["result"] With ACP, the same cross-framework delegation looks like this: Python from deepagents.middleware import AsyncSubAgentMiddleware, AsyncSubAgent # Any ACP-compliant server — regardless of what framework runs inside orchestrator = create_deep_agent( model="anthropic:claude-sonnet-4-6", middleware=[ AsyncSubAgentMiddleware(subagents=[ AsyncSubAgent(name="research", url="http://crewai-acp-server:8080"), AsyncSubAgent(name="coder", url="http://openai-acp-server:8081"), ]), ], ) The wrapping frameworks are invisible. The protocol standardizes streaming, cancellation, and structured error handling. Where Is This Heading? The inter-agent space is actively consolidating around a few patterns: ACP is the most explicit attempt at standardization. It bets that an open schema survives framework churn. You can swap the worker implementation without changing the orchestrator.Handoffs are winning on simplicity within the OpenAI ecosystem. For teams already on the OpenAI SDK, they're ergonomic and production-proven. The cross-framework limitation only matters if you leave the ecosystem.The agent-delegation model (Pydantic AI via agent-as-tool, AutoGen's event-driven redesign) is a better fit for peer networks where no single orchestrator coordinates everything.Framework-native protocols will likely remain dominant for the near term. Cross-framework standardization requires enough pain from fragmentation to motivate all players -> we're getting there, but not quite yet. If you're building a multi-agent system today and expect to stay within one framework, pick that framework's native protocol. If you're building infrastructure that needs to orchestrate agents across frameworks or you expect your team to evaluate multiple frameworks, investing in ACP-compatible interfaces from the start gives you the most flexibility.

Three connectivity features landed together this week, and they belong in one place because they build on each other. WebSockets moved into the core; the GraphQL client uses that same WebSocket support for subscriptions; and gRPC reuses the exact code-generation pattern GraphQL and OpenAPI already follow. This post is a tutorial for all three. By the end, you will have a live chat, a typed GraphQL client, and a typed gRPC client, and you will see how little code each one takes. These features come from PR #5133 (WebSockets) and PR #5141 plus PR #5099 (the typed clients). Part 1: WebSockets, No cn1lib Required WebSockets used to require the cn1-websockets cn1lib. They are now part of the framework as com.codename1.io.WebSocket, implemented natively on every port (a hand-rolled RFC 6455 handshake on JavaSE and Android, NSURLSessionWebSocketTask on iOS, the browser WebSocket on JavaScript), with no third-party dependencies pulled into your build. If you're using cn1-websockets you can keep using it. There's no change required from you. We moved the package up one level, so there's no conflict. Step 1: Open a Connection The new API is a final, fluent class with lambda handlers. You build it, attach handlers, and connect: Java // Good practice although in reality all current Codename One Platforms support WebSockets if (!WebSocket.isSupported()) { return; } WebSocket ws = WebSocket.build("wss://echo.example.com/socket") .onConnect(() -> Log.p("connected")) .onTextMessage(text -> addIncoming(text)) .onClose((code, reason) -> Log.p("closed " + code + " " + reason)) .onError(ex -> Log.e(ex)) .connect(); There is no URL-in-constructor subclassing trap from the old API; the connection is an object you hold. send(...) has a String and a byte[] overload, getReadyState() returns a WebSocketState, and close() does a clean close handshake. Step 2: Build the Chat Screen Here is a compact chat form. Outgoing messages are added immediately; incoming ones arrive on the onTextMessage handler, and because the handler can touch the UI we wrap that in callSerially: Java private WebSocket ws; private Container conversation; private void showChat(Form parent) { Form chat = new Form("Live Chat", BoxLayout.y()); conversation = chat.getContentPane(); TextField input = new TextField("", "Message", 20, TextField.ANY); Button send = new Button("Send"); send.addActionListener(e -> { String text = input.getText(); if (text.length() > 0 && ws != null) { ws.send(text); addBubble(text, true); input.clear(); } }); Container bar = BorderLayout.centerEastWest(input, send, null); chat.add(BorderLayout.SOUTH, bar); ws = WebSocket.build("wss://chat.example.com/room/general") .onTextMessage(text -> Display.getInstance() .callSerially(() -> addBubble(text, false))) .connect(); chat.show(); } private void addBubble(String text, boolean mine) { Label bubble = new Label(text); bubble.setUIID(mine ? "ChatBubbleMe" : "ChatBubbleThem"); Container line = FlowLayout.encloseIn(bubble); line.getStyle().setAlignment(mine ? Component.RIGHT : Component.LEFT); conversation.add(line); conversation.animateLayout(150); } That is a working real-time chat. The screen it produces, rendered in the simulator: Step 3: Negotiate a Subprotocol When You Need One If your server speaks a named subprotocol, set it during the handshake and read back what the server chose: Java WebSocket ws = WebSocket.build(url) .subprotocols("graphql-transport-ws") .onConnect(() -> Log.p("using " + ws.getSelectedSubprotocol())) .connect(); That graphql-transport-ws value is not an accident; it is exactly what the GraphQL subscriptions in the next part use. One reason to trust this implementation: our own screenshot CI now runs on it. The pipeline that ships rendered PNGs from each device back to the host machine uses a WebSocket as its transport, so the same code your app calls is carrying the binary payloads that validate the framework on every commit. Part 2: A Typed GraphQL Client cn1:generate-graphql turns a GraphQL schema into a typed client, and @GraphQLClient is the interface you write against. The runtime lives in com.codename1.io.graphql, and a GraphQLResponse<T> carries data and errors together so partial results survive. Step 1: Declare the Client Java @GraphQLClient("https://swapi.example.com/graphql") public interface StarWarsApi { @Query("query HeroName($episode: Episode) { hero(episode: $episode) { name homeworld { name } species { name } filmConnection { totalCount } } }") void hero(@Var("episode") Episode episode, OnComplete<GraphQLResponse<HeroData>> callback); @Subscription("subscription OnReview($ep: Episode!) { reviewAdded(episode: $ep) { stars } }") GraphQLSubscription onReview(@Var("ep") Episode ep, GraphQLSubscription.Handler<ReviewData> handler); static StarWarsApi of(String endpoint) { return GraphQLClients.create(StarWarsApi.class, endpoint); } } The build-time processor emits the implementation and a bootstrap that registers it; you never write the HTTP plumbing. The generator has two modes. The precise operations mode emits per-selection types from your operation documents; the schema-only quick-start mode auto-selects fields to a bounded depth (cn1.graphql.maxDepth). Step 2: Call It and Render the Result Java StarWarsApi api = StarWarsApi.of("https://swapi.example.com/graphql"); api.hero(Episode.EMPIRE, response -> { if (!response.isOk()) { return; } Container list = heroForm.getContentPane(); for (Hero h : response.getResponseData().heroes) { MultiButton row = new MultiButton(h.name); row.setTextLine2(h.homeworld + " . " + h.species); row.setUIID("HeroRow"); list.add(row); } heroForm.revalidate(); }); The list this populates, rendered in the simulator: Step 3: Subscriptions Ride the Core WebSocket A @Subscription returns a GraphQLSubscription backed by the core WebSocket using the graphql-transport-ws protocol from Part 1. New events arrive on the handler: Java GraphQLSubscription sub = api.onReview(Episode.JEDI, review -> Display.getInstance().callSerially(() -> showStars(review.stars))); // later sub.close(); This is the payoff of putting WebSockets in the core: the GraphQL layer did not need its own socket implementation; it just used the frameworks. Part 3: A Typed gRPC Client cn1:generate-grpc does the same trick for proto3. Point it at your .proto files and it emits hand-editable @ProtoMessage, @ProtoEnum, and @GrpcClient sources; the annotation processor generates the binary protobuf codecs and call sites into target/generated-sources so your source tree stays clean. There is no protoc dependency. Step 1: The Proto Java syntax = "proto3"; service Greeter { rpc SayHello (HelloRequest) returns (HelloReply); } message HelloRequest { string name = 1; } message HelloReply { string message = 1; } Step 2: Call the Generated Client Java GreeterGrpc g = GreeterGrpc.of("https://api.example.com"); HelloRequest req = new HelloRequest(); req.name = "world"; g.sayHello(req, "Bearer " + token, response -> { if (response.isOk()) { renderGreeting(response.getResponseData().message); } }); The wire protocol is gRPC-Web binary (application/grpc-web+proto), the standard variant for mobile and browser clients, which works with Envoy, the official grpcweb Go proxy, and the gRPC-Web filter in modern gRPC servers. Version one covers unary RPCs, all scalar types, nested messages, enums, and repeated fields; streaming, map<K,V>, well-known types, and import are out for now, and the parser errors cleanly when it meets one. Enums Bind Across All of It All three connectors share the build-time JSON and XML mapper, and that mapper now binds enums. Previously an enum field was treated as a nested reference, found no mapper, and silently did not serialize. It now writes with name() and reads with valueOf (unknown values decode to null), and it handles List<Enum>, across both JSON and XML. That is why the GraphQL Episode above is a real enum rather than a String, and it is a welcome fix for anyone using @Mapped directly. Keep Your Tokens Out of the Binary The gRPC and GraphQL samples pass a bearer token, so the rule bears repeating: never hard-code a token, and never check it into source or embed it in the app. Fetch it from your backend at runtime and store it with SecureStorage. A shipped binary can be unpacked, so anything baked into it is effectively public. These connectors learn from real specs. If a schema or a proto file does not generate the client you expected, please file an issue at github.com/codenameone/CodenameOne/issues with the source attached. The previous deep dive covered native Mac builds and desktop integration, and the release post has the full index. Tomorrow's post is the new advertising API.

By Shai Almog CORE

Modern applications deal with massive amounts of text — support tickets, CRM notes, blog posts, meeting transcripts, and internal documentation. The problem isn’t access to information anymore — it’s how quickly users can understand it. In our CRM system, we allow publishing long-form articles to a blog. However, users rarely want to read everything up front. To solve this, we introduced AI-powered summarization to generate short, readable previews. This improves: Content scanabilityUser engagementTime-to-information In this article, we consider: What text summarization isWhy AI summarization is powerfulSetting up OpenAI in RailsImplementing a summarization serviceBuilding a controller endpointHandling long documentsBackground processing with SidekiqReal-world use cases What Is Text Summarization? Text summarization is the process of condensing a large body of text into a shorter version while preserving its key information. There are two main approaches: 1. Extractive Summarization This selects the most important sentences directly from the original text. Example: Original: Ruby on Rails is a powerful web framework designed to make programming easier by favoring convention over configuration. Summary: Ruby on Rails is a web framework that simplifies development. 2. Abstractive Summarization This generates new sentences that capture the meaning of the text. This is where large language models like OpenAI shine. Why Use LLMs for Summarization? Traditional NLP methods struggle with context and nuance. OpenAI models can provide: Contextual understandingMulti-paragraph reasoningDomain adaptabilityNatural-sounding summaries This makes them ideal for summarizing in a project with different types of information processes: Blog postsDocumentsMeeting transcriptsCustomer feedbackKnowledge bases Setting Up OpenAI in a Rails project 1. First of all, install the OpenAI Ruby Gem. Add the gem: Ruby gem "ruby-openai" 2. Configure the API key. Add your API key to environment variables: Ruby export OPENAI_API_KEY="your_api_key" 3. Example initializer: Ruby OpenAI.configure do |config| config.access_token = ENV["OPENAI_API_KEY"] end Creating a Summarization Service In Rails, the best practice is to encapsulate OpenAI logic in a separate service object. Simple example: Ruby module Openai class Summarizer def initialize(text) @text = text @client = OpenAI::Client.new end def call response = @client.chat( parameters: { model: "gpt-4.1-mini", #or you can select another one messages: [ { role: "system", content: "You are a helpful assistant that summarizes text concisely." #you can define content with more detailed prompt }, { role: "user", content: "Summarize the following text:\\n\\n#{@text}" #also here define more detailed expected response } ], temperature: 0.3 } ) response.dig("choices", 0, "message", "content") end end end Possible roles: Plain Text Role Purpose system - > instructions for the model user - > input from the user assistant - > previous AI responses Usage: Ruby summary = Openai::Summarizer.new(article.content).call Example: Before and After Input (CRM article excerpt):Our platform allows teams to manage projects, track time, and generate reports across multiple departments... Output (AI summary): Centralized platform for project and time trackingSupports multi-department workflowsProvides reporting and analytics tools This summary can be shown in: Blog preview cardsTooltipsSearch results Example Controller Endpoint Now we expose this functionality via an API endpoint. Controller example: Ruby class Api::SummariesController < ApplicationController def create text = params[:text] summary = Openai::Summarizer.new(text).call render json: { summary: summary } end end Temperature controls randomness in the output. Plain Text 0.0 → deterministic 1.0 → very creative Additional Useful Parameters Also added parameters for improving summarization. 1. max_tokens: Limits the size of the generated response. Example: Ruby max_tokens: 200 #This prevents extremely long outputs. 2. top_pAlternative randomness control. Example: Ruby top_p: 0.9 #Usually you adjust temperature or top_p, not both 3. frequency_penalty discourages repeated phrases. Example: Ruby frequency_penalty: 0.2 #Useful when summaries become repetitive 4. presence_penalty encourages introducing new ideas. Example: Ruby presence_penalty: 0.1 #Not usually necessary for summarization, but can be used in specific tasks Prompt Engineering for Better Summaries and Why It Is Important The prompt design significantly impacts the output quality. Instead of a generic prompt: Plain Text "Summarize this text" Use structured instructions: Plain Text "Summarize the following text in 3 bullet points. Focus on the key ideas and avoid unnecessary details." This simple change improves clarity, consistency, and usefulness of the generated summary. In practice, prompt design becomes even more important when working with different types of content, such as technical documentation, CRM notes, etc. I wrote more about the features of Prompt Engineering on practical examples in my other article. Example of a more structured prompt: Ruby { role: "user", content: <<~PROMPT Summarize the following article in 5 bullet points. #{@text} PROMPT } Handling Very Long Documents LLMs have token limits, so large texts must be processed in chunks. Typical approach looks like this: Split text into chunksSummarize each chunkCombine summariesGenerate a final summary Avoid Naive Chunking This is not ideal: Ruby text.scan(/.{1,3000}/m) It may cut sentences in half. Prefer: Splitting by paragraphsSplitting by sentence boundaries Text Chunking Ruby class TextChunker def self.chunk(text) text.split("\\n\\n") end end Chunk Summarization Ruby def summarize_long_text(text) chunks = TextChunker.chunk(text) partial_summaries = chunks.map do |chunk| Openai::Summarizer.new(chunk).call end Openai::Summarizer.new(partial_summaries.join("\\n")).call end Using Background Jobs for Summarization Summarizing large text can take some time, so it’s better to process it asynchronously. Example of our service usage with Sidekiq: Generate Summary Job Ruby class GenerateSummaryJob include Sidekiq::Job def perform(article_id) article = Article.find(article_id) summary = Openai::Summarizer.new(article.content).call article.update!(summary: summary) end end Error Handling Always assume external APIs can fail. Ruby rescueStandardError=>e Rails.logger.error(e.message) fallback_summary end Also consider: RetriesTimeoutsMonitoring Cost Optimization When you use AI features in production, cost management becomes critical. depends primarily on token usage, meaning the more text you send and receive, the more you pay. Some tips for cost optimization that you need to know: 1. Limit Input Size The most effective optimization is reducing the amount of text sent to the AI model. Instead of summarizing an entire document, you can: Extract relevant sectionsSummarize those sections only Example filtering before sending to OpenAI: Ruby class TextPreprocessor MAX_LENGTH = 5000 def self.clean(text) text.strip[0...MAX_LENGTH] end end Usage: Ruby clean_text = TextPreprocessor.clean(article.content) summary = Openai::Summarizer.new(clean_text).call This ensures you never send extremely large inputs. 2. Choose the Right Model Not every task requires the most powerful model. For summarization, smaller models often perform well. Example: Ruby model: "gpt-4.1-mini" Advantages: Much cheaperFaster responsesGood summarization quality So, use larger models only for complex reasoning tasks. 3. Token Counting Before Requests Sometimes the text is larger than expected. Using a token estimation step helps prevent sending oversized prompts. Example: Ruby def too_large?(text) text.length > 12000 end If too large: chunk text into smaller chunkssummarize in parts (chunks) Conclusion Throughout this article, we built a summarization pipeline in Rails using a clean service-oriented approach: Simple summarization servicePrompt optimizationChunking for large documentsBackground processing with SidekiqCost and reliability improvements

When we are demoing an agentic product, it always looks clean and clear: the agent pauses, the human approves or rejects, and execution continues. But what happens when the human actually says no? Human-in-the-loop (HITL) sounds like a single feature. In practice, it covers a wide design space: Do you pause mid-execution or notify asynchronously? Is the human a peer agent or an external approver?Can the human edit the action, or only approve or reject it?Does the framework resume execution exactly where it paused, or is there anything else? These questions yield different answers across all major agent frameworks, and those answers have very real production consequences. I assumed that all frameworks would converge on a single pattern for HITL design, but I found them to be very different. This article compares the six frameworks and their implementations of HITL. What You Will Learn By the end of this article, you will be able to: Distinguish the three fundamental HITL patterns - durable graph interrupt, message-loop injection, and blocking gate, and know which framework implements each.Read working code for all six frameworks and understand the exact execution pause and how it resumes for the frameworks.Pick the right framework for your use case. The Fundamental Divide The three distinct HITL patterns can be described as Durable graph interrupt: In this pattern, the execution graph serializes the entire graph state and suspends at the exact node where approval was needed. Nothing happens until a decision is made. If the process exits, then it's saved in an external checkpointer, and the run resumes from the point of suspension. Message loop injection: In this pattern, there is no suspension as such. Humans act as a first-class participant in a multi-agent conversation, steering a reply like any other agent. The loop runs continuously, and the human response is just another round.Blocking gate/run-termination: In this pattern, the framework runs or ends the run cleanly at a designated point, either blocking in process until the caller responds or terminating and returning an approval pending object that the human needs to resolve before resuming. Resuming the run is the human's responsibility. frameworkpatterntrue suspensionhuman can edit actionresumable after process restartdeepagentsGraph interrupt (LangGraph)✓✓ approve / edit / reject✓AgnoHumanReview on Step/Loop✓Partial✗AutoGenUserProxy agent (message loop)✗✓ via messages ✗OpenAI Agents SDKneeds_approval interruptPartial✗PartialCrewAIstep_callback + human_input on Task✗✗✗Pydantic AIDeferred tools (requires_approval)Partial✗✗ deepagents + LangGraph: graph-level interrupt Installation: Shell pip install deepagents langgraph # Python >=3.10 required # Docs: https://docs.langchain.com/oss/python/deepagents/human-in-the-loop deepagents resume/interrupt sequence deepagents uses LangGraph's interrupt()primitive. When the model produces a tool call that requires approval, execution suspends at that exact graph node. The serialized state is stored via a LangGraph checkpointer; the process can exit entirely and resume hours later. Wiring Up the Middleware Python from deepagents import create_deep_agent from langchain.agents.middleware import HumanInTheLoopMiddleware, InterruptOnConfig hitl = HumanInTheLoopMiddleware( interrupt_on={ # True = approve / edit / reject all allowed "delete_file": True, # Restrict to approve/reject only, with static description "run_bash": InterruptOnConfig( allowed_decisions=["approve", "reject"], description="Review this shell command before execution", ), # Dynamic description generated from the tool call at runtime "send_email": InterruptOnConfig( allowed_decisions=["approve", "edit", "reject"], description=lambda tool_call, state, runtime: ( f"Approve sending email to: {tool_call['args'].get('to')}" ), ), } ) agent = create_deep_agent( model="anthropic:claude-sonnet-4-6", middleware=[hitl], ) What the Reviewer Sees (HITLRequest Structure) Python # Surfaced to the reviewer when delete_file is triggered { "action_requests": [ { "name": "delete_file", "args": {"path": "/workspace/output.log"}, "description": "Tool execution requires approval\n\nTool: delete_file\nArgs: ..." } ], "review_configs": [ { "action_name": "delete_file", "allowed_decisions": ["approve", "edit", "reject"] } ] } The Three Decision Types Python from langgraph.types import Command # Approve — run as-is graph.invoke( Command(resume={"decisions": [{"type": "approve"}]}), config={"configurable": {"thread_id": "session-123"}, ) # Edit — change args before running graph.invoke( Command(resume={ "decisions": [{ "type": "edit", "edited_action": { "name": "delete_file", "args": {"path": "/workspace/old-backup.log"} } }] }), config={"configurable": {"thread_id": "session-123"}, ) # Reject — agent receives explanation and stops retrying graph.invoke( Command(resume={ "decisions": [{ "type": "reject", "message": "Do not delete production logs. Archive instead." }] }), config={"configurable": {"thread_id": "session-123"}, ) Multi-Tool Batching If the model calls two tools in the same response, deepagents batches them into a single HITLRequest. One round-trip will handle both: Python # Single interrupt — two pending actions simultaneously graph.invoke(Command(resume={ "decisions": [ {"type": "approve"}, {"type": "reject", "message": "rm -rf is too broad — use a specific path"} ] })) AutoGen v0.4: the UserProxy pattern Installation: Shell pip install autogen-agentchat autogen-ext # Docs: https://microsoft.github.io/autogen/stable/ AutoGen models the human as a UserProxyAgent which is a peer participant in multi-agent conversation. There is no suspension. The loop runs continuously, and the human turn is when the proxy injects a message. AutoGen message-loop HITL Python from autogen_agentchat.agents import AssistantAgent, UserProxyAgent from autogen_agentchat.teams import RoundRobinGroupChat from autogen_agentchat.conditions import TextMentionTermination from autogen_ext.models.openai import OpenAIChatCompletionClient assistant = AssistantAgent( "assistant", model_client=OpenAIChatCompletionClient(model="gpt-4o"), system_message=( "You are a helpful agent. Always describe what you are about to do " "and ask for confirmation before executing file operations." ), ) # input_func is called when the proxy needs human input # Replace `input` with an async queue for web applications user_proxy = UserProxyAgent("human", input_func=input) team = RoundRobinGroupChat( participants=[assistant, user_proxy], termination_condition=TextMentionTermination("DONE"), ) await team.run(task="Clean up old log files in /tmp") Limitation: The conversation loop never truly suspends. If the process exits mid-conversation, the state is lost. For async web UIs, you'd need a background thread and an asyncio queue to bridge human input. There's no built-in checkpointing. Agno: HumanReview on Steps Installation: Shell pip install agno # Docs: https://docs.agno.com/reference/workflows/step Agno's HITL uses a HumanReview config object attached to workflow steps. It supports confirmation gates before execution, user input collection, and post-execution output review: Python from agno.workflow import Workflow, Step from agno.workflow.types import HumanReview from agno.agent import Agent from agno.models.anthropic import Claude extract_agent = Agent(name="Extractor", model=Claude(id="claude-haiku-4-5"), ...) transform_agent = Agent(name="Transformer", model=Claude(id="claude-haiku-4-5"), ...) load_agent = Agent(name="Loader", model=Claude(id="claude-sonnet-4-6"), ...) workflow = Workflow( name="DataPipeline", steps=[ Step(name="extract", agent=extract_agent), Step(name="transform", agent=transform_agent), Step( name="load", agent=load_agent, # Pause and require human confirmation before this step runs human_review=HumanReview(requires_confirmation=True), ), Step( name="verify", agent=load_agent, # Pause after execution for a human to review the output human_review=HumanReview(requires_output_review=True), ), ], ) HumanReview fields: fieldscopewhat it doesrequires_confirmationStep, Loop, Router, ConditionPause before the step executesconfirmation_messageStep, Loop, Router, ConditionCustom prompt shown to the reviewerrequires_user_inputStep, RouterCollect freeform user input before continuingrequires_output_reviewStep, RouterPause after execution; accepts bool or Callable[[StepOutput], bool] for conditional reviewrequires_iteration_reviewLoop onlyReview after each loop iterationon_rejectAllOnReject.skip (default), cancel, or retry (re-run the step with human feedback)on_errorAllOnError.pause triggers HITL on step failure - human decides retry or skiptimeout / on_timeoutAllTimeout in seconds; on_timeout is cancel (default), skip, or approve Resumability: Agno has no workflow-level checkpoint equivalent to LangGraph's checkpointer. If the process exits while a step is awaiting human input, the workflow state is lost. Resumability requires external session storage wired by the caller. OpenAI Agents SDK: needs_approval interrupt Installation: Shell pip install openai-agents # Docs: https://openai.github.io/openai-agents-python/ The OpenAI Agents SDK uses a needs_approval parameter on function_tool. When set, the run loop pauses and surfaces a ToolApprovalItem that the caller approves or rejects via RunState: Python from agents import Agent, function_tool, Runner @function_tool(needs_approval=True) def delete_file(path: str) -> str: """Delete a file at the given path.""" import os os.remove(path) return f"Deleted {path}" # needs_approval can also be a callable for conditional approval @function_tool( needs_approval=lambda ctx, args, call_id: args.get("path", "").startswith("/prod") ) def write_file(path: str, content: str) -> str: """Write content to a file.""" with open(path, "w") as f: f.write(content) return f"Wrote {path}" agent = Agent( name="FileAgent", instructions="Help the user manage files.", tools=[delete_file, write_file], ) async def run_with_approval(): result = await Runner.run(agent, "Delete the old backup file") if result.interruptions: # Convert result to a resumable state, then resolve each pending approval state = result.to_state() for item in result.interruptions: print(f"Approve {item.raw_item.name}({item.raw_item.arguments})? [y/N]: ", end="") if input().strip().lower() == "y": state.approve(item) else: state.reject(item, rejection_message="User rejected this action") # Resume: pass the mutated state back to Runner result = await Runner.run(agent, state=state) print(result.final_output) Limitation: The approval flow is approve-or-reject only. There's no structured "edit" decision type. Humans cannot modify tool arguments through the SDK's approval mechanism. Partial cross-restart resumability is available via state.to_string() / RunState.from_string() and the human is responsible for persisting and restoring the serialized state externally. CrewAI: step_callback + human_input Installation: Shell pip install crewai # Docs: https://docs.crewai.com/en/concepts/crews CrewAI has two distinct mechanisms with very different semantics. step_callback — Observational Only step_callback fires after each agent step and receives an AgentAction | AgentFinish object. It cannot block or modify the next step: Python from crewai import Agent, Crew, Task from crewai.agents.crew_agent_executor import AgentAction, AgentFinish def review_step(step: AgentAction | AgentFinish) -> None: if isinstance(step, AgentAction): print(f"Tool used: {step.tool}, input: {step.tool_input}") elif isinstance(step, AgentFinish): print(f"Agent finished: {step.return_values}") researcher = Agent( role="Researcher", goal="Research the topic", backstory="An expert researcher.", verbose=True, ) task = Task(description="Research quantum computing trends", agent=researcher) crew = Crew( agents=[researcher], tasks=[task], step_callback=review_step, ) crew.kickoff() human_input — Blocking Task-Output Review Setting human_input=True on a Task does produce a real synchronous pause. After the agent finishes its work for that task, execution blocks on input() , and the human can provide free-form feedback before the output is finalized: Python from crewai import Agent, Crew, Task researcher = Agent( role="Researcher", goal="Research the topic", backstory="An expert researcher.", verbose=True, ) task = Task( description="Research quantum computing trends and summarize findings.", expected_output="A summary of the latest quantum computing developments.", agent=researcher, human_input=True, # blocks after agent finishes, before output is accepted ) crew = Crew(agents=[researcher], tasks=[task]) crew.kickoff() # Agent completes its work, then execution pauses: # > Please provide feedback on the agent's output (or press Enter to accept): Key distinction from tool-call-level gates: human_input fires after the agent has already finished the task and all tool calls have already executed. You are reviewing the output, not approving individual actions before they run. The human provides free-form text feedback. There is no structured approve/edit/reject schema, no async queue support, and no state serialization. Because it calls input() directly, it blocks the calling thread, and it is incompatible with async web servers (FastAPI, Starlette) without bridging to a separate thread and queue. Pydantic AI: Deferred Tools Installation: Shell pip install pydantic-ai # Docs: https://pydantic.dev/docs/ai/tools-toolsets/deferred-tools/ Pydantic AI has a first-class HITL primitive called Deferred Tools. Mark a tool with requires_approval=True (or raise ApprovalRequired conditionally) and the agent run terminates with a DeferredToolRequests object instead of a final answer. The caller resolves approvals and resumes with the original message history. Pydantic AI deferred-tool approval sequence Declaring Tools That Require Approval Python from pydantic_ai import Agent from pydantic_ai.exceptions import ApprovalRequired agent = Agent("anthropic:claude-sonnet-4-6") # Always requires approval @agent.tool(requires_approval=True) async def delete_file(ctx, path: str) -> str: import os os.remove(path) return f"Deleted {path}" # Conditional: only requires approval for destructive commands @agent.tool async def run_bash(ctx, command: str) -> str: import subprocess risky = ["rm", "drop", "truncate"] if any(r in command for r in risky): raise ApprovalRequired(metadata={"reason": "destructive command detected"}) return subprocess.check_output(command, shell=True).decode() Handling DeferredToolRequests and Resuming Python from pydantic_ai.tools import DeferredToolRequests, DeferredToolResults, ToolDenied async def run_with_approval(): result = await agent.run("Delete the old backup file") if isinstance(result.output, DeferredToolRequests): approvals = {} for tool_call in result.output.approvals: print(f"Approve {tool_call.tool_name}({tool_call.args})? [y/N]: ", end="") if input().strip().lower() == "y": approvals[tool_call.tool_call_id] = True else: # ToolDenied lets you pass a custom message back to the model approvals[tool_call.tool_call_id] = ToolDenied( message="User rejected this action - do not retry." ) # Resume: pass original message history + approval decisions result = await agent.run( message_history=result.all_messages(), deferred_tool_results=DeferredToolResults(approvals=approvals), ) print(result.output) Inline resolution with HandleDeferredToolCalls For cases where you want to resolve approvals within the same run (e.g., a CLI prompt that doesn't need to persist state), use the HandleDeferredToolCalls capability: Python from pydantic_ai.capabilities import HandleDeferredToolCalls from pydantic_ai.tools import DeferredToolRequests, DeferredToolResults, ToolDenied async def interactive_approver(ctx, requests: DeferredToolRequests) -> DeferredToolResults: approvals = {} for tool_call in requests.approvals: print(f"Approve {tool_call.tool_name}({tool_call.args})? [y/N]: ", end="") if input().strip().lower() == "y": approvals[tool_call.tool_call_id] = True else: approvals[tool_call.tool_call_id] = ToolDenied(message="Rejected by user.") return DeferredToolResults(approvals=approvals) agent = Agent( "anthropic:claude-sonnet-4-6", tools=[delete_file, run_bash], capabilities=[HandleDeferredToolCalls(interactive_approver)], ) Limitation: There is no durable state serialization. If the process exits between the first run (which returns DeferredToolRequests) and the resume, the run cannot be recovered. The caller(human) must persist result.all_messages() and the pending tool call IDs externally. There is also no structured "edit" decision type; the human can approve or deny, but cannot modify tool arguments through the SDK. Choosing the Right Pattern use casebest fit Long-running agent, async human reviewer with durable resume deepagents only Human needs to edit tool args before execution deepagents only Step-level gates with on_reject retry loops, no durable resume Agno HumanReview Conversational co-pilot, real-time back-and-forth AutoGen Approve/reject specific tools, run stays in-process OpenAI Agents SDK Approve/reject specific tools, run terminates for async handling Pydantic AI Deferred Tools Audit logging, no blocking needed CrewAI step_callback Review task output after agent finishes (not tool-call level) CrewAI human_input=True Conclusion There is no universal answer to HITL in agent frameworks. The right choice depends on three questions before choosing your framework: at what granularity does a human need to intervene (tool call, step, or task output), whether the reviewer responds in real time or hours later, and whether you need the process to survive a restart between the interrupt and the resume. If the answer to any of the last two is "yes," deepagents with a LangGraph checkpointer is the only framework that handles both today. For everything else, the landscape is richer than it first appears: Pydantic AI's Deferred Tools give you structured tool-call-level approval without a graph runtime; Agno gives you powerful step-level gates with retry semantics; and OpenAI Agents SDK gives you the simplest possible approve/reject path when you control the process lifecycle. The mistake most teams make is treating HITL as an afterthought. The primitives each framework exposes are not interchangeable, and switching from an observational callback to a durable interrupt requires rearchitecting the execution model, not just swapping a parameter. The decision tree above is meant to surface that choice before it becomes expensive to undo.

In a microservices system, that tight coupling turns a small hiccup into a cascading slowdown. Thread pools fill, retries amplify traffic, and suddenly your simple request is blocked on half the fleet. My executive summary: asynchronous messaging with Kafka helps systems keep moving when individual components inevitably slow down or fail. It does this by decoupling producers from consumers, absorbing traffic spikes, and allowing services to evolve without tying their availability directly to one another. Code Patterns in Spring Boot With Kafka Spring for Apache Kafka gives me two primitives that feel pleasantly old Spring KafkaTemplate for sending and @KafkaListener for receiving. That template/listener model is intentionally similar to other Spring integration tech, which keeps application code focused on domain logic instead of raw client plumbing. Below is a compact (but production-shaped) pattern: externalized config via @ConfigurationProperties, a service port for publishing, a REST command endpoint, a consumer with a real error strategy (DLT), and a REST error advice. Java // === Messaging config (externalized, type-safe) === @ConfigurationProperties(prefix = "messaging.orders") @Validated record OrdersMessagingProps( @NotBlank String topic, @NotBlank String dltTopic ) {} // === DTO (event contract) === public record OrderCreatedEvent(UUID orderId, UUID userId, BigDecimal total, Instant createdAt) {} // === Service port (keeps domain testable, Kafka swappable) === public interface OrderEventPublisher { void publishOrderCreated(OrderCreatedEvent event); } // === Adapter: Kafka producer === @Component class KafkaOrderEventPublisher implements OrderEventPublisher { private final KafkaTemplate<String, OrderCreatedEvent> template; private final OrdersMessagingProps props; KafkaOrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> template, OrdersMessagingProps props) { this.template = template; this.props = props; } @Override public void publishOrderCreated(OrderCreatedEvent event) { // Keying by orderId keeps per-order ordering and drives partitioning decisions. template.send(props.topic(), event.orderId().toString(), event); } } // === REST command API (synchronous edge, async core) === @RestController @RequestMapping("/v1/orders") class OrdersController { private final OrderService orderService; // domain port OrdersController(OrderService orderService) { this.orderService = orderService; } @PostMapping public ResponseEntity<Map<String, Object>> create(@Valid @RequestBody CreateOrderRequest req) { UUID orderId = orderService.create(req.userId(), req.total()); // persists + publishes event return ResponseEntity.accepted().body(Map.of("orderId", orderId, "status", "ACCEPTED")); } record CreateOrderRequest(@NotNull UUID userId, @NotNull @Positive BigDecimal total) {} } // === Domain service port (implementation can use outbox, transactions, etc.) === public interface OrderService { UUID create(UUID userId, BigDecimal total); } // === Consumer: downstream service reacts to events === @Component class BillingListener { @KafkaListener(topics = "${messaging.orders.topic}", groupId = "${spring.kafka.consumer.group-id}") void onOrderCreated(OrderCreatedEvent event) { // Idempotency belongs here: process-by-key + store processed eventId/orderId to avoid duplicates. // Do work (charge card, create invoice, etc.) } } // === Kafka consumer error handling: retries + DLT === @Configuration class KafkaErrorHandlingConfig { @Bean DefaultErrorHandler defaultErrorHandler(KafkaTemplate<Object, Object> template, OrdersMessagingProps props) { var recoverer = new DeadLetterPublishingRecoverer(template, (rec, ex) -> new TopicPartition(props.dltTopic(), rec.partition())); // Backoff and retry policy are configurable; keep it finite to avoid poison-pill loops. return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3)); } } // === REST error handling (ProblemDetail) === @RestControllerAdvice class ApiErrors { @ExceptionHandler(IllegalArgumentException.class) @ResponseStatus(HttpStatus.BAD_REQUEST) ProblemDetail badRequest(IllegalArgumentException ex) { var pd = ProblemDetail.forStatusAndDetail(HttpStatus.BAD_REQUEST, ex.getMessage()); pd.setTitle("Invalid request"); return pd; } } A few been-burned-before notes on the code above. Spring Kafka’s reference docs are explicit that KafkaTemplate is the convenience wrapper for producing, and DefaultErrorHandler + DeadLetterPublishingRecoverer is a first-class way to route failed records to dead-letter topics after retries. If we want non-blocking retries, Spring Kafka also provides @RetryableTopic, which orchestrates retry topics and a DLT automatically useful when transient failures are common and you want predictable retry delay semantics. Containers and Local Dev With Docker Compose When I’m chasing down event flow bugs, I like local environments that feel like the old days: one command, deterministic startup order, and no mystery dependencies. Docker Compose is still the quickest way to stand up Kafka alongside your services, and Confluent publishes straightforward Docker-based tutorials and compose examples for running Kafka locally. For the service image itself, multi-stage builds are the modern classic compile in a builder stage, and copy the artifact into a slimmer runtime stage. Docker documents multi-stage builds as a way to reduce the final image contents and keep build dependencies out of production. Dockerfile # Multi-stage Dockerfile for a Spring Boot service (orders-service) FROM eclipse-temurin:21-jdk AS build WORKDIR /workspace COPY mvnw pom.xml ./ COPY .mvn .mvn RUN ./mvnw -q -DskipTests dependency:go-offline COPY src src RUN ./mvnw -q -DskipTests package FROM eclipse-temurin:21-jre WORKDIR /app COPY --from=build /workspace/target/*.jar app.jar EXPOSE 8080 ENTRYPOINT ["java","-jar","/app/app.jar"] And here’s a Compose file that wires up Kafka and Schema Registry, plus an example Spring Boot service. The exact image choices are illustrative. Your production choices are unspecified and should reflect your standards and security posture. YAML # compose.yaml (local/dev) services: zookeeper: image: confluentinc/cp-zookeeper:7.6.0 environment: ZOOKEEPER_CLIENT_PORT: 2181 kafka: image: confluentinc/cp-kafka:7.6.0 depends_on: [zookeeper] ports: ["9092:9092"] environment: KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 schema-registry: image: confluentinc/cp-schema-registry:7.6.0 depends_on: [kafka] ports: ["8081:8081"] environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092 orders: build: ./orders-service depends_on: [kafka] ports: ["8080:8080"] environment: SPRING_KAFKA_BOOTSTRAP_SERVERS: kafka:9092 MESSAGING_ORDERS_TOPIC: orders.events MESSAGING_ORDERS_DLTTOPIC: orders.events.dlt SCHEMA_REGISTRY_URL: http://schema-registry:8081 Deploying on Kubernetes or AWS On AWS, the Kafka decision is usually managed or self-managed. If you choose Amazon MSK, the cluster lives in your VPC, pick subnets across distinct Availability Zones, and connect clients using the cluster’s bootstrap brokers. That’s the networking baseline, and it’s not optional. MSK is VPC-first by design. For authentication/authorization, MSK supports IAM access control. AWS documents the client configuration for IAM mechanisms. In EKS, I typically pair MSK IAM with IRSA so pods can obtain AWS credentials the AWS way, while ECS services would use task roles instead. Both patterns are documented by AWS, and your choice here is unspecified. Kubernetes service discovery is usually the easy part. Services and Pods get DNS names so workloads can call each other by name rather than IP. Kafka itself is reached via bootstrap broker endpoints or via internal Services, but either way, you want the strings in externalized config, not hardcoded. Here’s a minimal Kubernetes Deployment/Service for a Kafka client service. Values like region, account IDs, and MSK endpoints are unspecified placeholders. YAML apiVersion: apps/v1 kind: Deployment metadata: name: orders namespace: apps spec: replicas: 2 selector: matchLabels: { app: orders } template: metadata: labels: { app: orders } spec: serviceAccountName: orders-sa # IRSA-bound (role ARN unspecified) containers: - name: orders image: <UNSPECIFIED_AWS_ACCOUNT_ID>.dkr.ecr.<UNSPECIFIED_REGION>.amazonaws.com/orders:<TAG> ports: [{ containerPort: 8080 }] env: - name: SPRING_KAFKA_BOOTSTRAP_SERVERS value: "<UNSPECIFIED_MSK_BOOTSTRAP_BROKERS>" - name: MESSAGING_ORDERS_TOPIC value: "orders.events" - name: MESSAGING_ORDERS_DLTTOPIC value: "orders.events.dlt" readinessProbe: httpGet: { path: /actuator/health/readiness, port: 8080 } initialDelaySeconds: 10 --- apiVersion: v1 kind: Service metadata: name: orders namespace: apps spec: selector: { app: orders } ports: - port: 80 targetPort: 8080 Operationally, MSK exposes metrics into CloudWatch (AWS/Kafka), and broker logs can be delivered to CloudWatch Logs (or S3/Firehose). That combination gives you the classic visibility loop: throughput, lag, under-replicated partitions, and error logs without running your own monitoring plane. For distributed tracing in async flows, OpenTelemetry is my default vocabulary now. Spring Boot supports OpenTelemetry export via OTLP, and OpenTelemetry defines Kafka semantic conventions so your producer/consumer spans and attributes stay consistent across tools. CI/CD and the Hard-Earned Field Notes For CI/CD, I keep it boring: build once, push an immutable image, deploy via a declarative mechanism. AWS Prescriptive Guidance provides a clear GitHub Actions pattern for building Docker images and pushing to Amazon ECR, which is a solid baseline when your region/account is unspecified until configured. YAML # .github/workflows/orders.yml name: orders on: push: branches: ["main"] jobs: build_push_deploy: runs-on: ubuntu-latest permissions: id-token: write contents: read steps: - uses: actions/checkout@v4 - uses: actions/setup-java@v4 with: distribution: temurin java-version: "21" - name: Build & test run: ./mvnw -q test package - name: Configure AWS credentials (OIDC) uses: aws-actions/configure-aws-credentials@v4 with: role-to-assume: arn:aws:iam::<UNSPECIFIED_AWS_ACCOUNT_ID>:role/<UNSPECIFIED_GHA_ROLE> aws-region: <UNSPECIFIED_REGION> - name: Login to ECR run: | aws ecr get-login-password --region <UNSPECIFIED_REGION> \ | docker login --username AWS --password-stdin <UNSPECIFIED_AWS_ACCOUNT_ID>.dkr.ecr.<UNSPECIFIED_REGION>.amazonaws.com - name: Build & push image run: | IMAGE=<UNSPECIFIED_AWS_ACCOUNT_ID>.dkr.ecr.<UNSPECIFIED_REGION>.amazonaws.com/orders:${{ github.sha } docker build -t $IMAGE ./orders-service docker push $IMAGE - name: Deploy to EKS (example) run: | aws eks update-kubeconfig --name <UNSPECIFIED_EKS_CLUSTER> --region <UNSPECIFIED_REGION> kubectl -n apps set image deploy/orders orders=$IMAGE Now, the part I wish someone had handed me in 2016: Kafka gives you strong tools, but it does not remove distributed-systems truths. You still need safeguards on the consumer side: idempotent processing, disciplined schema management, and clearly defined retry and dead-letter topic behavior. Kafka’s documentation is careful about the limits of “exactly once” guarantees. Idempotent producers and transactions can strengthen delivery semantics, but achieving true end-to-end exactly-once behavior, especially when external side effects are involved, still depends on deliberate system design. For schema governance, Kafka itself doesn’t ship a schema registry, but acknowledges third-party registries; in practice, Confluent Schema Registry and Apicurio Registry are common choices. Both store schemas out-of-band, so messages carry only a schema identifier, and both support evolvable contracts across Avro/JSON Schema/Protobuf depending on your ecosystem. Conclusion and Best Practices If you take one lesson from my legacy brain into modern event-driven systems, let it be this: asynchrony is a reliability feature, not a performance trick. Kafka’s durable log and consumer group model decouples uptime and absorbs spikes, but you only get the real benefit when you treat schemas as contracts, consumers as idempotent processors, and failure handling as first-class application behavior. On AWS, the operational baseline is non-negotiable. MSK lives in your VPC across AZ subnets, clients connect via bootstrap brokers, IAM auth is configured explicitly, and observability lives in CloudWatch. Do those fundamentals early, and Kafka stops feeling like a mysterious black box and starts feeling like the dependable workhorse it was built to be.

Our focus was all over the place this week with work that targeted many different directions: desktop, monetization, communication, media, and more. This fits with our roadmap of one platform that delivers the promise Java never delivered: WORA for Everything Everywhere. But before we dig into the new features, there's one number I'm particularly proud of… Open Issues Are Under 350 The open issue count is now below 350 (332 at the moment of this writing). That is the result of a deliberate pass through the tracker, closing things that were already fixed, reproducing the ones that were not, and fixing a batch of them outright. Some of the reports we closed this week had been open since 2015. We got a little side-tracked in the process (it is hard to read an old report without wanting to fix it on the spot), but the direction is set: we want this number to keep dropping, and by a lot. We went over the issue tracker, starting with the oldest issues and working our way back. You may recall that when we started tracking this, going under the 500-issue mark was a major milestone, and that was just a few weeks back! What Shipped Every one of the bigger items has its own deep-dive tutorial. Here is the tour, with the links to the full posts. Your App Is Now a Native Mac App Your existing Codename One app can ship as a 100% native Mac app today, with zero porting effort. No rewrite, no bundled JVM, no Electron shell: the project you already have produces a lean native Mac binary the same way it produces your iPhone app, on the same Metal renderer and battle-tested native pipeline. And it arrives feeling like a real Mac app, not a phone in a window: native title bar, native menu bar, interactive scrollbars, and desktop notifications come with it. Two of this week's features in one shot: the sample below uses the new advertising API covered later in this post, running as a native Mac app from the same Java code that produces the iOS and Android builds: The full tutorial, including the new desktop menu and shortcut APIs and the Mac signing hints, is in Your Codename One App, Now A Native Mac App. WebSockets in the Core com.codename1.io.WebSocket is now part of the framework with no cn1lib required, implemented natively on every port. A live connection is a fluent one-liner: Java WebSocket ws = WebSocket.build("wss://chat.example.com/room/general") .onConnect(() -> Log.p("connected")) .onTextMessage(text -> Display.getInstance() .callSerially(() -> addBubble(text, false))) .connect(); This is the foundation the GraphQL subscriptions below ride on, and it is trusted enough that our own screenshot CI uses it as the transport for device renders. The hands-on tutorial that builds a live chat with it is WebSockets, gRPC, and GraphQL In The Core. A Typed GraphQL Client From Your Schema cn1:generate-graphql turns a GraphQL schema into a typed client: you declare an interface against your operations and the build emits the implementation, with zero HTTP plumbing on your side. A @Subscription gets you live server-pushed updates over the new core WebSocket: Java @GraphQLClient("https://swapi.example.com/graphql") public interface StarWarsApi { @Query("query HeroName($episode: Episode) { hero(episode: $episode) { name } }") void hero(@Var("episode") Episode episode, OnComplete<GraphQLResponse<HeroData>> callback); @Subscription("subscription OnReview($ep: Episode!) { reviewAdded(episode: $ep) { stars } }") GraphQLSubscription onReview(@Var("ep") Episode ep, GraphQLSubscription.Handler<ReviewData> handler); } Note that Episode is a real enum: enum binding landed in the JSON/XML mapper alongside this work, fixing a long-standing gap for @Mapped users too. The full tutorial is in WebSockets, gRPC, And GraphQL In The Core. gRPC Clients With No Protoc cn1:generate-grpc does the same for proto3. Point it at your .proto files and it generates the messages, the binary protobuf codecs, and the client, speaking the standard gRPC-Web wire protocol that works with Envoy and modern gRPC servers. There is no protoc to install and no native dependency; calling a service looks like this: Java GreeterGrpc g = GreeterGrpc.of("https://api.example.com"); HelloRequest req = new HelloRequest(); req.name = "world"; g.sayHello(req, "Bearer " + token, response -> { if (response.isOk()) { renderGreeting(response.getResponseData().message); } }); Together with cn1:generate-openapi and cn1:generate-graphql, this means a typed client for practically any backend is one Maven goal away. The walkthrough from proto file to running call is in WebSockets, gRPC, And GraphQL In The Core. A New Advertising API Advertising support had quietly rotted across three dead-end legacy mechanisms. A pluggable, format-complete com.codename1.ads subsystem replaces all of them, with a modern AdMob reference provider, GDPR consent and iOS App Tracking Transparency built in, and a simulator placeholder provider so you can exercise every format without a device. A rewarded ad, the format people most often ask about, is now this short: Java RewardedAd ad = new RewardedAd("your-rewarded-ad-unit-id"); ad.setAdListener(new AdListener() { public void onLoaded() { ad.show(reward -> grantCoins(reward.getType(), reward.getAmount())); } }); ad.load(); The full story, including banners, native ads, app-open ads, and the provider SPI, is in A New Advertising API, Built From The Ground Up. Background Execution and Push Constraint-based background work, foreground services, push topics, shared-content handling, and a much richer local notification API, all of it usable in the simulator so you can debug these flows on your desktop. You describe what the work needs, not when to poll: Java WorkRequest req = WorkRequest.builder("daily-sync", SyncWorker.class) .setRequiresNetwork(true) .setRequiresCharging(true) .setPeriodic(6 * 60 * 60 * 1000L) .build(); BackgroundWork.schedule(req); The walkthrough, from progress notifications and inline replies to push topics and shared content, is Background Work, Push Topics, and Richer Notifications. Building Screens From Screenshots, and a Simpler Initializr Generated apps ship with a codename-one agent skill under .claude/skills/, so an AI agent working in your project already knows how to build, test, and screenshot a Codename One UI. PR #5161 teaches it the single most common design task: "make this screen look like this mockup." The hard part of that task was that the agent had no objective measure of how close it had gotten. The existing screenshot test only compares a render to a baseline the system produced itself, which measures consistency, not correctness. The new CompareToMockup tool is a single-file, pure-JDK CLI that scores a render against a target image and prints a similarity percentage: a STRUCTURAL score (an SSIM-style perceptual measure, robust to font and anti-aliasing noise against a vector mockup) and a PIXEL score (the fraction of pixels within the framework's own three-channel "same pixel" tolerance). It has a region mode so device chrome does not sabotage the score, a --diff heatmap, and a --min gate. That gives the agent a real signal: render, score, read the heatmap, adjust, repeat until the number stops climbing. A companion DesignImport tool turns a Figma, Sketch, or Adobe XD file (and, after PR #5168, the tokens.css from an HTML or React mockup) into a starter theme.css so the agent adjusts rather than starting from a blank page. The skill can also update itself now through an UpdateSkills tool, so a project generated months ago can pull the current guidance instead of carrying a frozen copy. Agents can now automatically update the skills to the latest versions and also describe the content of Codename One GUIs. This is valuable as they review their work and don't need to use vision, which is both more expensive and not as accurate. We also added the ability to check component alignment, which is often a problem that LLMs find difficult. There is also a new linter that I think we should expose to the human developers as well in the future. Right now, you can see all of these tools and use them just as an agent would, but they are more CLI-oriented. PR #5168 also rebuilt the Initializr, the tool that scaffolds a new project, around the Codename One design language, and trimmed it so it is easier to approach. It leads with the essentials (main class, package, and a Java or Kotlin toggle) and tucks IDE, localization, Java version, and current settings into collapsible cards, with a live preview and a single generate bar at the bottom. The four-template picker became the Java/Kotlin toggle, and the accent, rounded-buttons, and custom-CSS controls were dropped. The project model behind it is unchanged, so generated projects are the same; this is purely about lowering the barrier to getting started. Smaller Fixes Worth Knowing About Several of these came straight out of the issue-tracker pass: cubic-bezier() motion now matches CSS. PR #5122 fixes #1524: Motion.createCubicBezierMotion was feeding its control points into a 1D polynomial directly, so the curve did not match the CSS cubic-bezier() it was modeled on. It does now so animations might act differently in some cases.Always-tensile on the X axis. PR #5112 closes #1399, the next-oldest open issue (filed March 2015): setAlwaysTensile(true) now applies horizontally as well as vertically. This means you would see the rubber-band effect also on X axis scrolling.Validation highlights on tap-away. PR #5123 closes #1459: a field with an invalid value is now highlighted when you tap into a different field, not only when you press the virtual keyboard's next/enter.EncodedImage.dispose() actually frees memory. PR #5127 makes dispose() release the decoded image and the encoded bytes (it was a no-op before) and adds isDisposed(). Closes #3733.NetworkManager.ping(). PR #5130 adds ping(url, timeoutMillis), a real server-reachability probe to pair with the device-side isConnected(). Closes #3669.ImageViewer drag bubbling. PR #5132: a vertical drag on an ImageViewer at zoom 1 now scrolls the parent container instead of being swallowed. Closes #3700. This means you can now include many image viewers in a scrollable Y container.Graphics.isVisible(). PR #5129 adds a clip-intersection primitive so a zoomed canvas can cull off-screen content and skip the decode/scale. Closes #3846.Screenshot block now covers peers. PR #5107: ios.blockScreenshotsOnEnterBackground=true was hiding the render surface but leaving peer components (such as a BrowserComponent's WKWebView) visible in the app-switcher snapshot. Fixed.Better site search. PR #5090 sorts the on-site search index newest-first and stops the giant developer-guide page from crowding out every result. We also have dates visible next to blog post results in the search, and a new highlight explaining we don't search the developer guide/Javadoc. A Note on Contributions We stopped accepting community pull requests. We want to be precise about why, because it would be easy to read more into this than is there. This is not about AI-generated PRs. We are not policing how a contribution was written. The real reason is mechanical: our CI does not run correctly against pull requests from forks. The screenshot pipeline, the device runners, and the protocol tests all need credentials and a setup that a forked PR cannot get, so a community PR cannot actually be validated by the same gates we hold our own work to. This isn't something we can easily solve without introducing major security vulnerabilities to our process. Recently, a change that looked completely safe slipped through and triggered a CI regression that took the builds down. That was on us, not on the contributor, but it convinced us that merging code we cannot fully run is the wrong trade. So the door is open the other way. Please keep filing issues. A clear issue with a test case is genuinely no trouble for us to pick up, and as the number above shows, we are actively working on the tracker. If you have a fix in mind, describe it in the issue, attach the failing case, and we will carry it through CI ourselves. To everyone who has sent us patches over the years, and especially the people who contributed recently: thank you. The effort was real and appreciated, and this decision is about our pipeline, not your work. Upcoming Attractions Four deep-dive tutorials follow this one, one per day: Saturday. Native Mac builds and deeper desktop integration. PRs #5053, #5136, #5170.Sunday. WebSockets, gRPC, and GraphQL in the core. PRs #5133, #5099, #5141.Monday. The new advertising API. PR #5169.Tuesday. Background work, push topics, and richer notifications. PR #5142. Wrapping Up The issue tracker is here, and it is the best place to reach us right now. The discussion forum is here, and the Build Cloud console is at /console/. The Playground, Initializr, and Skin Designer are where they have always been.

By Shai Almog CORE

The xAI Grok API provides access to powerful frontier models, including the Grok 4 series, supporting chat completions (text + vision), image generation, tool calling (function calling and built-in tools like web search), and more advanced features. Quick Intro Sign up at https://x.ai/api.Generate an API key from the console.Install pip install xai-sdk.Set env var: export XAI_API_KEY="your_key_here".Models list: https://docs.x.ai/developers/models. I'll share some samples in Python. Learn how to use Grok AI - xAI Basic Chat API Call Let's first prepare our project before making the API call 1. Install the xai-sdk. Shell pip install xai-sdk 2. Set env var: export XAI_API_KEY="your_key_here" or use .env file. Now, create a new file and this basic setup: Python import os from xai_sdk import Client from xai_sdk.chat import user, system from dotenv import load_dotenv load_dotenv() XAI_API_KEY = os.environ.get("XAI_API_KEY") client = Client(api_key=XAI_API_KEY) Ensure you can print out your XAI_API_KEY correctly at this stage. Next, let's call the chat function: Python ... model = "grok-4-1-fast-non-reasoning" chat = client.chat.create(model=model) chat.append(system("You are Grok, a highly intelligent, helpful AI assistant.")) chat.append(user("How can I be a good developer?")) response = chat.sample() print(response.content) Feel free to switch the model based on your needs or preferences. Here is an example output: Grok AI API basic call Image Generation API Let's see how to generate an image with Grok API. We'll need to use the "grok-imagine-image" model for this. Python ... response = client.image.sample( model="grok-imagine-image", prompt="detective cat searching on website" ) print(f"Generated image: {response.url}") The output is a URL like this: Image generation API using xAI API Video Generation API Generating a video is as easy as generating an image with Grok API. We'll need to use the "grok-imagine-video" model for this. Python response = client.video.generate( prompt="A glowing crystal-powered rocket launching from the red dunes of Mars, ancient alien ruins lighting up in the background as it soars into a sky full of unfamiliar constellations", model="grok-imagine-video", duration=10, aspect_ratio="16:9", resolution="720p", ) print(response.url) Grok Video API example You can set the duration, aspect ratio, and resolution. Tools in Grok The xAI Grok API features powerful tool-calling capabilities, allowing Grok to go far beyond simple text generation. It can take real actions such as performing web searches, running code, retrieving information from your own data sources, or invoking any custom functions you've defined. From x.ai - available tools Tool Calling (Function Calling) Let's start by calling a custom function, as it'll help us call any internal or external API or function. Let's say we want to call a function to look for an item's price. First, we need to define the function, such as adding the name, description, and parameters. Python ... import json from xai_sdk.chat import user, tool, tool_result ... # Define tools tools = [ tool( name="get_item_price", description="Get the price of an item from the store", parameters={ "type": "object", "properties": { "item_name": {"type": "string", "description": "Name of the item to get the price for"}, }, "required": ["item_name"] }, ), ] Upon calling the client method, we now need to include the tool we declared above. Python chat = client.chat.create( model="grok-4.20-reasoning", tools=tools, ) chat.append(user("What is the price of a laptop?")) response = chat.sample() print("========= response ===========") print(response) print("==========================") Important: At this stage, Grok doesn't care if we have the actual function to check the price or not. The AI simply wants to know "what tools are available" for them to use. Try to run the code to see the output from the chat call. Function calling output sample As you can see, Grok can detect the tool we need to call. You can see it from outputs > message > tool_calls . It consists of the name of the function and the arguments that are extracted from the user's prompt, so it'll be dynamic. Function Call Simulation Next, let's create a fake function to call. In real life, it could be a call to a database or APIs. Python def get_item_price(item_name): prices = { "laptop": 999.99, "smartphone": 499.99, "headphones": 199.99, } return {"item_name": item_name, "price": prices.get(item_name, "Item not found")} Following up on the latest code, we can check if the response has a "tool_calls" object or not. If so, we'll call the actual function we just declared above. Python # Handle tool calls if response.tool_calls: chat.append(response) for tc in response.tool_calls: args = json.loads(tc.function.arguments) result = get_item_price(args["item_name"]) chat.append(tool_result(json.dumps(result))) response = chat.sample() print(response.content) We need to loop through the tool_calls objectWe need to extract the argument to pass to the functionCall the actual function alongside the argument valueAdd the information back to our chat method Now, calling the chat.sample() method, will include all the information we received from calling the "fake function" before. Sample result for function calling Let's try with a different prompt: Shell chat.append(user("I need to buy two laptops and a smartphone. Can you tell me how much that will cost?")) Here is the result: Function calling result sample Web Search API Grok can access real-time information through this feature, so you can get up-to-date content. Unlike the function calling above, we don't need to declare a custom function, as it's an internal tool. Here is a simple example: Python import os from xai_sdk import Client from xai_sdk.chat import user from xai_sdk.tools import web_search from dotenv import load_dotenv load_dotenv() XAI_API_KEY = os.environ.get("XAI_API_KEY") client = Client(api_key=XAI_API_KEY) chat = client.chat.create( model="grok-4.20-reasoning", # reasoning model tools=[web_search()], include=["verbose_streaming"], ) chat.append(user("Grok VS OpenAI API")) is_thinking = True for response, chunk in chat.stream(): for tool_call in chunk.tool_calls: print(f"\nCalling tool: {tool_call.function.name} with arguments: {tool_call.function.arguments}") if response.usage.reasoning_tokens and is_thinking: print(f"\rThinking... ({response.usage.reasoning_tokens} tokens)", end="", flush=True) if chunk.content and is_thinking: print("\n\nFinal Response:") is_thinking = False if chunk.content and not is_thinking: print(chunk.content, end="", flush=True) print("\n\nCitations:") print(response.citations) Use tools=[web_search()]To show what's happening in the process, we use include=["verbose_streaming"],is_thinking variable is to check if the process is still running (a boolean variable) Web Search API with Grok AI As you can see, it'll perform several searches on the internal database with different queries. It'll then visit a specific URL after that to get more context. Allowed Domains You can search only in specific domains using allowed_domains. Python tools=[ web_search(allowed_domains=["grokipedia.com"]), ], Exclude Domains Vice versa, you can exclude specific domains: Python chat = client.chat.create( model="grok-4.20-reasoning", tools=[ web_search(excluded_domains=["grokipedia.com"]), ], ) Better Web Search API While you can specifically choose the domain, the keyword Grok uses to find answers on the internet is random. For example, when I'm asking for "Top 3 pizza restaurants from Google Maps in Boston. Share some reviews and ratings for each place." This is what I saw from the thinking process: It needs to perform multiple queries before returning the answer. Another sample, when asking simply for three images: It runs across multiple pages, and unfortunately, the links are not valid. Grok may hallucinate at this point. Web Search API Alternative In some cases, AI-generated keywords are fine, but if you're building an app where you want efficiency and full control over the process, the native "Web Search Tool" can be replaced with a simple API call to a specific API your app needs. For example, to find answers online, SerpApi offers 100+ APIs. Need a generic Google answer? We have: Google Search APIGoogle AI OverviewGoogle AI Mode Same with Bing, DuckDuckGo, and other top search engines. Need a restaurant review? We have: Yelp Reviews APIGoogle Maps Reviews API Need an API for traveling apps? We have: Google Hotels APIGoogle Flights APITripAdvisor API and more! See how SerpApi is the Web Search API for your AI apps, LLM, and agents. Using Grok API With SerpApi To get a sense of how SerpApi works, feel free to test the results in our playground. You can play with different parameters and directly see the JSON sample we return. SerpApi Playground Sample Case Let's say we want to find images via Google Image API like this: Sample result search with SerpApi Step 1: Preparation You can register for free at serpapi.com to get your API key. Step 2: Parsing Keyword Let's say we need three images from Google. Since users can type anything, we need to parse the keyword, as SerpApi simply performs a search using a particular keyword. Python USER_QUERY = "Show me 3 cute cat images from the internet" # Step 1: Ask Grok to extract a search keyword from the user's natural language keyword_chat = client.chat.create(model="grok-3-fast") keyword_chat.append(system("Extract the most relevant search keyword or phrase from the user's message. Reply with only the keyword, nothing else.")) keyword_chat.append(user(USER_QUERY)) keyword_response = keyword_chat.sample() search_keyword = keyword_response.content.strip() print(f"Extracted keyword: {search_keyword}") Step 3: Search via SerpApi We now have the keyword. Let's run a search on SerpApi. Python # Step 2: Search via SerpAPI using simple requests (Google Images) serpapi_params = { "api_key": SERPAPI_API_KEY, "engine": "google_images", "q": search_keyword, "hl": "en", "gl": "us", } serpapi_url = "https://serpapi.com/search" serpapi_response = requests.get(serpapi_url, params=serpapi_params) results = serpapi_response.json() At this stage, you already have the answers you're looking for. Step 4: Filter Results (Optional) Sometimes, we don't need all the information. It's good to filter it programmatically first, so we don't use too many tokens. For example, I'm only interested in the top five answers: Python image_results = results.get("images_results", [])[:5] formatted_results = "\n".join( f"- {img.get('title', 'No title')}: {img.get('original', img.get('thumbnail', 'No URL'))}" for img in image_results ) print(f"\nSerpAPI results:\n{formatted_results}") We can also format the answer as a bonus. Step 5: Reply in Natural Language (Optional) Depending on your application, you may want to answer the user back in natural language. We just need to pass the answers above back to the AI: Python # Step 3: Feed results back to Grok for a final response final_chat = client.chat.create(model="grok-3-fast") final_chat.append(system("You are a helpful assistant. Use the provided search results to answer the user's question.")) final_chat.append(user(f"User question: {USER_QUERY}\n\nSearch results from SerpAPI:\n{formatted_results}\n\nPlease answer the user's question based on these results.")) final_response = final_chat.sample() print(f"\nFinal Response:\n{final_response.content}") Final result: You can try the other APIs for other use cases. Sidenote It's also possible to call the API with the OpenAI SDK. Sample: Python from openai import OpenAI client = OpenAI( api_key=os.getenv("XAI_API_KEY"), base_url="https://api.x.ai/v1", ) Check out the full SerpAPI article collection here.

Managing high-volume message traffic in distributed architectures is crucial. Efficient use of database and CPU resources is also very important. There are structures that allow us to receive messages in batches. The default Spring Kafka "BatchMessageListener" structure addresses this need. However, the processing of these messages often goes through a sequential bottleneck. This article will discuss the structure and usage of Kotlin Coroutines in detail. We will examine how to maximize Kafka message processing performance using Structured Concurrency principles and Resource Throttling techniques. Architectural Bottleneck: Sequential I/O Blocking On the current Kafka listener: Database or external service calls made for each message directly increase total processing times. If the processing speed of a message lags behind the message arrival speed and the max-poll-interval-ms time is exceeded, the consumer is removed from the consumer group. Rebalancing is triggered, and the partitions of that consumer are redistributed to other consumers in the group. Kotlin @KafkaListener(topics = ["usage-pool-topic"]) fun usagePoolListener(records: List<ConsumerRecord<String, String>>) { records.forEach { record -> processRecord(record) // Network latency + DB I/O blocking } } Solution 1. Batch-Fetch and In-Memory Map Structure Before any concurrent code is entered, data is retrieved collectively from all necessary entities. Multiple separate queries are converted into a batch query before data processing begins. The N+1 query problem is solved at the application layer. All data is cached once before being broken down into concurrent operations. Having the data cached significantly reduces our reliance on the database. Using the associateBy function, we transform the data into a map structure with X access times. This allows us to read the data safely from the maps instead of reading each concurrent operation from the database. Kotlin val messages = records.map { objectMapper.readValue(it.value(), UsagePoolRecord::class.java) } val usagePoolEntities = usagePoolRepository .findByIds(messages.map { it.usagePoolId.toBigInteger() }) .associateBy { it.usagePoolId } val lockEntities = lockRepository .findByUserIds(messages.map { it.userId }) .associateBy { it.userId } 2. Structured Concurrency Memory Management With Chunking The chunk structure serves two purposes. It prevents the creation of coroutines simultaneously. This prevents unnecessary memory usage. Each chunk writes to the database after all coroutines have completed their operations. Unnecessary connection pool consumption is avoided. Kotlin messages.chunked(150).forEach { chunk -> // Each chunk of 150 records is processed concurrently } Resource Isolation With limitedParallelism Why limitedParallelism? If the database connection pool has, for example, X connections, keeping the parallelism limit below X prevents "Connection Timeout" errors. Kotlin messages.chunked(150).forEach { chunk -> val deferredResults = chunk.map { record -> CoroutineScope(Dispatchers.IO.limitedParallelism(15)).async { try { processRecord(record, usagePoolEntities, lockEntities) } catch (e: Exception) { log.error("Operation error: ${record.key()}", e) buildErrorRecord(record, e) } } } val results = deferredResults.awaitAll() // Structural waiting collectAndAggregate(results) } The Dispatchers.IO.limitedParallelism(X) command limits the number of concurrent coroutines to X, preventing the DB connection pool from being exhausted.Each coroutine returns a result with the async command. The awaitAll() command waits for all coroutines in the chunk to finish before proceeding to the next step. runBlocking This function blocks callers until all concurrent operations are complete. This is the correct approach here because: It ensures that the Kafka consumer remains blocked to maintain its offset commit structure until all records in the batch are processed. We still benefit from concurrent operation parallelism within the runBlocking block. 3. Thread-Safe Result Structure After the awaitAll() operation, all results are collected in thread-safe queues. Then a single batch write operation takes place. Using MutableList structures to combine results returned from parallel processed coroutines can lead to data loss. At this point, lock-free data structures should be preferred. ConcurrentLinkedQueue uses CAS (Compare-And-Swap) algorithms instead of synchronized blocks. This provides superior performance in high-content write operations. Why Shouldn't We Use ConcurrentLinkedQueue? Concurrent operations (concurrent functions) perform simultaneous write operations to a shared collection of results. Using MutableList leads to race conditions. It performs well in secure and concurrent write operations. Kotlin data class AggregatedRecords( val processedSave: ConcurrentLinkedQueue<ProcessedEntity> = ConcurrentLinkedQueue(), val toDelete: ConcurrentLinkedQueue<UsagePoolEntity> = ConcurrentLinkedQueue(), val retryQueue: ConcurrentLinkedQueue<RetryEntity> = ConcurrentLinkedQueue() ) The DataIntegrityViolationException return is important. When two consumer instances are processing the same record, one of them falls into a unique constraint violation. Instead of making the entire batch fail, record-by-record deletion is performed. Kotlin AggregatedRecords.processedSave .chunked(150) .forEach { batch -> try { processedRepository.saveAll(batch) } catch (e: DataIntegrityViolationException) { batch.forEach { record -> try { processedRepository.save(record) } catch (e: DataIntegrityViolationException) {} } } } 4. Error Tolerance in Write Operations Batch write (saveAll) operations are performant. However, a "Unique Constraint" error in a single record can cause the entire batch to fail. The following structure is critical to meet Optimistic Locking or Idempotency requirements. Kotlin aggregatedRecords.processedSave.chunked(150).forEach { batch -> try { processedRepository.saveAll(batch) } catch (e: DataIntegrityViolationException) { // Fallback: Try one by one if batch fails batch.forEach { record -> try { processedRepository.save(record) } catch (innerException: DataIntegrityViolationException) { log.warn("Duplicate record skipped: ${record.id}") } } } } 5. Data Flow Diagram Ingress: The Kafka batch is caught with runBlocking.Preparation: All necessary context data is retrieved bulk from the DB.Execution: Coroutines are started asynchronously in chunks.Synchronization: The completion of all coroutines is awaited as a barrier point with awaitAll().Egress: Collected results are made permanent with saveAll. Performance Analysis and Results Conclusion Processing Kafka messages in Spring Boot with Kotlin Coroutines not only increases speed but also improves code readability and makes resource management deterministic (predictable). The use of runBlocking allows us to build a bridge between the blocking Kafka consumer thread and the suspended world without disrupting Kafka's offset management mechanism. Dependencies XML <dependency> <groupId>org.jetbrains.kotlinx</groupId> <artifactId>kotlinx-coroutines-core</artifactId> <version>1.7.3</version> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>

From McKinsey to BCG, from the EU to ISO, and from vendors and consulting firms worldwide, everyone has their own version of the AI Readiness Assessment. A quick search will turn up dozens of them, and they arrive in my feeds daily. Some are cursory and superficial, some granular and thoughtful. Some are a series of random questions, while others are carefully categorized into elements like strategy, data, technology, talent, governance, and culture. Some can be completed in an hour, while others require extensive preparation, discovery, and participation. They all have flaws common to these sorts of instruments: they are self-reported, self-scored, and self-interpreted. All pollsters know that self-reported data is inherently suspect. Pollsters and researchers have a name for what happens when people assess themselves without any external calibration: they call it bias, and it’s been studied exhaustively. All of the cognitive biases that affect people’s responses to events also show up in survey responses. Once a respondent can see how the questions map to the results and how their responses might be perceived, the instrument stops being a diagnostic and becomes a negotiation between the user and the instrument: “How can I position myself to look my best, highlight my value, or compete with the next guy…” When the scores come back, confirmation bias creeps in, urging us to accept the results we like and reject those we don’t. I’m not suggesting self-assessments are worthless; I use them with my clients frequently. As an advisor to clients thinking through their preparedness for the AI transition, I’ll often run a facilitated session in which an assessment instrument is the foundation. When thoughtfully applied, with realistic expectations about what a self-assessment can and cannot do, they can spark a useful conversation. They have great value to me as a neutral ‘forcing function’, requiring my clients to dig deeper into the reality of their circumstances without my direct challenge, but instead through the application of a ‘standards’ instrument. Let’s explore the options for published assessments or maturity models in the AI readiness space. The Landscape The most rigorous frameworks come from standards bodies. The NIST AI Risk Management Framework (AI RMF) organizes AI governance across four functions — Govern, Map, Measure, Manage — and is widely regarded as the most coherent and flexible foundation available. It’s an excellent educational and internal management tool, providing a detailed “how-to guide” that organizations can adapt to their specific context. NIST AI RMF is an example of the structured, granular, and comprehensive assessment that regulated industries and hyperscale corporations, with their complex legal and regulatory exposures, might apply. ISO/IEC 42001 is another example of the disciplined approach taken by international standards bodies in the AI arena. Like many ISO standards, they offer certifications that attest to one’s ability to apply and interpret their assessments. Probably the most precise and strategic of the standards-based surveys, preparation for an enterprise to gain ISO 42001 certification is a multi-month commitment, not a self-assessment you run in a conference room. The payoff is high credibility in the enterprise’s AI safety and integrity. McKinsey, Deloitte, and BCG each publish maturity models and readiness frameworks, typically built around six dimensions: StrategyDataTechnologyTalentGovernance Culture These instruments are often well-constructed and genuinely informative. Designed at enterprise scale, targeted towards the potential clients of these consulting giants, they are marketing tools as much as they are survey instruments, calibrated to prepare potential clients for the next engagement. A regional retailer, mid-market professional services firm, or specialized financial advisory practice will frequently find that the questions don’t map to their operational reality and that the scores that they derive don’t carry much useful information for their circumstances. What Self-Assessment Can Do A well-constructed assessment process, applied thoughtfully, can expose the enterprise's capabilities and gaps and can aid in the prioritization of AI efforts. It has a few important properties beyond its primary use as a ‘scorecard.’ First, it forces structured reflection. Many organizations are deploying AI somewhere, perhaps in an experimental or pilot program, without having formally considered the strategic or operational concerns that will arise as they scale. A credible assessment process creates the occasion to ask them: Do we have a policy for AI-generated client communications? Who has decision rights and accountability for actions AI may take? Have we considered the cultural, educational, and organizational changes we may need to apply?Do we have a governance process to set guardrails and standards? Second, it builds cross-functional alignment. Walking a leadership team through a common instrument often surfaces the fact that the COO, the compliance officer, and the practice leads have fundamentally different assumptions about AI risk and AI opportunity. Surfacing it early, in a structured context, is valuable regardless of where the assessment scores land. Third, it establishes a baseline. A self-assessment conducted today, repeated in six months, gives you directional data about whether your AI transition strategy is effective, whether the policies you wrote are being operationalized, and whether the training you delivered is changing behavior. That longitudinal function is where the trends become visible. What It Cannot Do Organizations, like individuals, often overrate their own capabilities and underrate their own exposure, not from dishonesty, but due to the limits of self-perception. The team filling out the assessment is the same team that built the systems being assessed. They share the same blind spots, the same assumptions, the same organizational memory, and norms. An instrument that asks “Do you have an AI governance policy?” will return “yes” from organizations whose policy is a two-paragraph statement no one has read, and from organizations with a mature, enforced framework. Only the conversation, not the one-word response, can uncover the real workings of the enterprise. There is also a scale problem. The major instruments were built for either government-scale complexity (NIST, ISO) or enterprise-scale exercises (the big-firm maturity models). A self-assessment produced without rigor or fit produces a slide deck that reassures, and perhaps misleads, leadership, while the real gaps are only exposed during the transition. For organizations without a basis of understanding about AI’s potential, or the cultural ability to have an honest self-improvement conversation, a 200-question framework has the potential to become a low-value quagmire. How to Do This Well Clients often come to me with a pre-selected assessment in hand, either to proudly show me their outstanding readiness score or to express concern over their obvious gaps. Many clients want to walk through an assessment with their teams before they engage an advisor. For those, I recommend a few guidelines: Match the instrument to your context. A small wealth management firm and a multinational manufacturer have different AI risk profiles, different regulatory exposures, and different resource constraints. The NIST AI RMF, applied at full depth, is probably not your starting point if you don’t have a risk management function to operationalize it. A consulting firm maturity model targeted at enterprise transformation may not reflect your operational reality. Calibrate the tool to the right level of complexity for your organization, at the right level of discipline and resource allocation, to ensure AI efficacy and safety for your unique circumstances. Scale your discipline to your evolving maturity. Run it cross-functionally, not in the IT department. AI transition is a business problem with a technology dimension, not a technology problem with business implications. The assessment should include legal, compliance, operations, and leadership, and the answers from each function should be compared analytically, not averaged into a meaningless score. Rights, roles, responsibilities, and gaps exposed must be engaged. Be honest about what the score means. A score is a prompt for conversation, not a certification. The process, not the score, is where the value resides. Low scores are more valuable than high marks because they illuminate the potential areas of friction or unpreparedness. If this becomes a competitive exercise, either between departments or organizations, you’re doing it wrong. Treat the assessment as a beginning, not a conclusion. The output of a self-assessment should be a prioritized set of actions. The questions that surface about data governance, accountability structures, and AI in workflows that haven’t been formally reviewed are the starting point for the deeper, more valuable diagnostic work that inevitably follows.

The MovieManager project has been updated to use JDK 25 and the AOT cache from project Leyden. Project Leyden is part of the OpenJDK project and provides cached linking and cached performance statistics. That means the time spent linking at startup is moved to build time, and the statistics are created during a test run at build time as well. Because of that, the JVM loads the needed classes already linked and starts compiling the hot code paths immediately. The MovieManager application starts in less than half the time with these optimizations without any code changes. All these advantages come with preconditions: Exactly the same JVM version at build time, training time, and run timeThe same OS(Linux is used here) and libc at all steps -> (No Alpine-based Docker Images)Same CPU architecture, for example, AMD64 or ARM64 The steps to use Project Leyden: Build the Spring Boot ApplicationExtract the Spring Boot ApplicationDo a training run with the extracted Application to create the AOT cacheCreate the Docker Image with the extracted Application and the AOT cache Building and Training the Application The first step is to build the Spring Boot JAR. The MovieManager project has an integrated build that builds the Angular frontend and the Spring Boot backend with this Maven command: Shell ./mvnw clean install -Ddocker=true -Dnpm.test.script=test-chromium Project Leyden does not support Spring Boot Jars. The Jar has to be extracted to help Project Leyden find the used library jars of the project. To do that, this command needs to be used: Shell java -Djarmode=tools -jar backend/target/moviemanager-backend-0.0.1-SNAPSHOT.jar extract --destination extracted The result is the directory ‘extracted’ with the application jar and a sub-directory ‘lib’ that contains the used libraries. The second step is to create the AOT cache. To do that, the application has to run in production conditions. That means using a real PostgreSQL database with the database driver. That enables the JDK to record all the needed classes of the project and to create realistic performance statistics for the code compilation. To do this, a PostgreSQL database has to be started(done here in a Docker container), and the Application has to do the full startup. These commands are needed: Shell docker pull postgres:13 docker run --name local-postgres -e POSTGRES_PASSWORD=sven1 -e POSTGRES_USER=sven1 -e POSTGRES_DB=movies -p 5432:5432 -d postgres java -XX:+UseG1GC -XX:MaxGCPauseMillis=50 -XX:+UseCompressedOops -XX:+UseCompactObjectHeaders -XX:+ExitOnOutOfMemoryError -XX:MaxDirectMemorySize=64m -XX:+UseStringDeduplication -Xlog:aot -XX:AOTCacheOutput=app.aot -Dspring.context.exit=onRefresh -Djava.security.egd=file:/dev/./urandom -jar extracted/moviemanager-backend-0.0.1-SNAPSHOT.jar --spring.profiles.active=prod The Java command runs the application with the parameter ‘-Dspring.context.exit=onRefresh’ that makes Spring Boot do the full startup and exit then. The parameters ‘-Xlog:aot -XX:AOTCacheOutput=app.aot’ enable the logging of the AOT process and the creation of the ‘app.aot’ that is the AOT cache. The AOT cache contains everything that is needed for a fast startup of the application. If the AOT cache should also contain information to improve production performance, it would have to start up and process realistic production requests. That is beyond the scope of this article. The third step is to test the new application setup: Shell java -XX:+UseG1GC -XX:MaxGCPauseMillis=50 -XX:+UseCompressedOops -XX:+UseCompactObjectHeaders -XX:+ExitOnOutOfMemoryError -XX:MaxDirectMemorySize=64m -XX:+UseStringDeduplication -Xlog:class+path=info -XX:AOTCache=app.aot -Xlog:aot -Djava.security.egd=file:/dev/./urandom -jar extracted/moviemanager-backend-0.0.1-SNAPSHOT.jar --spring.profiles.active=prod The start-up time of the new setup with the AOT cache can be compared to the start-up time of the Spring Boot jar. On a medium-powered laptop, the times are: 9 seconds for the Spring Boot Jar3.5 seconds for the new setup with the AOT cache Creating a Docker Image To use the application in production, it needs to be packaged into a Docker image. The Docker image needs to contain the extracted application setup and the AOT cache. The base image needs to have the exact same JDK version, OS, and the same libc. That means small base images like Alpine cannot be used. The created Image can not be small because it contains 180 MB of AOT cache and a larger base image. This can be done with this Dockerfile: Dockerfile FROM eclipse-temurin:25.0.3_9-jdk-jammy WORKDIR /application ARG JAR_FILE=extracted/*.jar COPY ${JAR_FILE} moviemanager-backend-0.0.1-SNAPSHOT.jar COPY extracted/ ./ COPY app.aot app.aot ENV JAVA_OPTS="-XX:+UseG1GC \ -XX:MaxGCPauseMillis=50 \ -XX:+UseCompressedOops \ -XX:+UseCompactObjectHeaders \ -XX:+ExitOnOutOfMemoryError \ -XX:MaxDirectMemorySize=64m \ -XX:+UseStringDeduplication" ENTRYPOINT exec java $JAVA_OPTS -XX:+AOTClassLinking \ -XX:AOTCache=app.aot \ -Xlog:class+path=info \ -Djava.security.egd=file:/dev/./urandom \ -jar moviemanager-backend-0.0.1-SNAPSHOT.jar It copies the new application setup in the image and adds the AOT cache. The name of the application jar is in the AOT cache and has to be exactly the same as during the creation of the AOT cache. The ‘JAVA_OPTS’ also have to be the same. If the JDK version in the build environment changes, the version of the base image has to be adjusted accordingly. The parameter ‘-Xlog:class+path=info’ makes analyzing AOT problems much easier. The Docker container size is 705 MB. That makes the container about double the size of a Docker container with a Spring Boot Jar and an Alpine-based JDK image. Creating a Build Pipeline Creating Docker images for an application by hand is unsustainable in a production environment. A build pipeline is needed. The MovieManager project is hosted on GitHub; because of that, the project uses a GitHub Workflow as a build pipeline. The complete code for the build pipeline is in the script. The steps of the GitHub pipeline can be recreated in other environments too. The first step is to set up the PostgreSQL database service to be used in this build: YAML jobs: analyze: name: Analyze runs-on: ubuntu-latest env: POSTGRES_URL: jdbc:postgresql://localhost:5432/movies services: postgres: image: postgres:latest env: POSTGRES_USER: sven1 POSTGRES_PASSWORD: sven1 POSTGRES_DB: movies ports: - 5432:5432 options: >- --health-cmd="pg_isready -U sven1 -d movies" --health-interval=10s --health-timeout=5s --health-retries=5 The commands set up the PostgreSQL service in the build pipeline with user, password, dbname, and dbport. The ‘POSTGRES_URL’ is set to access the database later. The second step is to check out the project: YAML steps: - name: Checkout repository uses: actions/checkout@v3 It checks out the contents of the master branch. The third step is to provide the JDK: YAML - name: Setup Java JDK uses: actions/setup-java@v3 with: distribution: 'temurin' java-version: 25 JDK version 25 is the minimum to use the project Leyden with linking and performance statistics. The fourth step builds the Spring Boot Jar: YAML - name: Build with Maven if: matrix.language == 'java' run: | ./mvnw clean install -Ddocker=true That is the Maven command to build the project. The fifth step is to find the Spring Boot jar: YAML - name: Find fat jar if: matrix.language == 'java' id: jar run: | JAR_PATH=$(find ./backend/target -type f -name "*SNAPSHOT.jar" | head -n 1) echo "Found JAR: $JAR_PATH" echo "jar=$JAR_PATH" >> $GITHUB_OUTPUT The sixth step is to extract the Spring Boot jar: YAML - name: Unpack fat jar if: matrix.language == 'java' id: UNPACK run: | java -Djarmode=tools -jar ${{ steps.jar.outputs.jar } extract --destination extracted EXTRACTED_PATH=$(find . -type d -name "extracted" | head -n 1) echo "Found directory: $EXTRACTED_PATH" echo "extracted=$EXTRACTED_PATH" >> $GITHUB_OUTPUT The seventh step is to get the name of the extracted application jar: YAML - name: find extracted jar if: matrix.language == 'java' id: EXTRACT run: | EXTRACTED_JAR=$(find "${{ steps.UNPACK.outputs.extracted }" -type f -name "*.jar" | head -n 1) EXTRACTED_JAR=${EXTRACTED_JAR#./} echo "Found extracted JAR: $EXTRACTED_JAR" echo "extracted=$EXTRACTED_JAR" >> $GITHUB_OUTPUT The eighth step is to create the AOT cache: YAML - name: Create AOT cache if: matrix.language == 'java' id: AOT env: JAVA_TOOL_OPTIONS: "" _JAVA_OPTIONS: "" JDK_JAVA_OPTIONS: "" run: | EXTRACTED_JAR="${{ steps.EXTRACT.outputs.extracted }" echo "jar=$EXTRACTED_JAR" echo "JAVA_TOOL_OPTIONS=$JAVA_TOOL_OPTIONS" echo "_JAVA_OPTIONS=$_JAVA_OPTIONS" echo "JDK_JAVA_OPTIONS=$JDK_JAVA_OPTIONS" JAVA_OPTS="-XX:+UseG1GC -XX:MaxGCPauseMillis=50 -XX:+UseCompressedOops -XX:+UseCompactObjectHeaders -XX:+ExitOnOutOfMemoryError -XX:MaxDirectMemorySize=64m -XX:+UseStringDeduplication" java $JAVA_OPTS \ -XX:+AOTClassLinking \ -XX:AOTCacheOutput=app.aot \ -Xlog:aot \ -Dspring.context.exit=onRefresh \ -Dspring.datasource.url="${{ env.POSTGRES_URL }" \ -Dspring.profiles.active=prod \ -jar "$EXTRACTED_JAR" || echo "AOT Training finished with exit code $?" This runs the application startup with the PostgreSQL database to create the AOT cache. The ninth step shows the exact JDK version used in the AOT cache generation: YAML - name: Show Jdk version if: matrix.language == 'java' id: JDK run: | JDK_VERSION=$(java -version 2>&1) VERSION=$(echo "$JDK_VERSION" | sed -n 's/.*build \([^[:space:]]*\)-LTS.*/\1/p') echo "JDK_VERSION=$JDK_VERSION" echo "VERSION=$VERSION" MY_VERSION="jdk=$VERSION" In case of problems with using the AOT cache. The first check is the version shown here against the JDK version in the Docker base image. The tenth step creates the Docker image: YAML - name: Build and push uses: docker/build-push-action@v6 if: matrix.language == 'java' with: context: . file: ./Dockerfile build-args: | JAR_PATH=${{ steps.EXTRACT.outputs.extracted } LIB_PATH=${{ steps.aot.outputs.extracted } push: false tags: angular2guy/moviemanager:latest This step can push the Docker image to an image repository. Conclusion The results of using the AOT cache of project Leyden are impressive. Cutting the startup time in half without any code change is amazing. The effort to create the AOT cache and set up the new application is a one-time investment. The impact of the larger Docker Images is low. That makes scaling application instances in Kubernetes clusters up and down much more flexible because the time to the availability of a new application instance is much lower. In Kubernetes environments with scaling of application instances, the AOT cache is a significant step forward and should be used. For serverless applications 3.5 seconds startup time is too slow. Their project, CrAC or Native Image, would be needed. Project CrAC needs code changes and testing. Native Image has the closed-world assumption, which makes it hard to prove that larger applications work correctly. Alternatives are Node.js with Nest.js and TypeScript, or Go with its libraries. Project Leyden is not finished in JDK 25. There are plans to add compiled code to the AOT cache in the future. The JVM is an impressive piece of technology that is still improving further.