--- title: "Build Java agents" description: "Run agents as durable stages with typed submission, tools, approvals, conversation stores and controlled provider configuration." category: "Java SDK" --- The agent runtime supplies planning, thinking, tool execution, confirmation, review and response stages. Your application configures a model and tools; the same worker facade owns transport and execution lifecycle. [Download the agent demo](/downloads/pipelogiq-java-agent-demo.zip). It includes an offline harness, a continuously running agent worker and a separate typed producer. Start with [deterministic testing](/docs/java/testing), then use a real provider when you want to test model behavior. ## Install the agent and testing modules From the matching SDK source root: ```sh ./mvnw -B -pl sdk-runtime,sdk-agent,sdk-testing -am install -DskipTests ``` An agent application adds `pipelogiq-sdk-agent` alongside the recommended runtime dependency: ```xml com.pipelogiq pipelogiq-sdk 0.5.0 com.pipelogiq pipelogiq-sdk-agent 0.5.0 ``` `pipelogiq-sdk-testing` is optional for production. The downloadable demo includes it at normal scope so its offline `main` can run. In an application test suite, use test scope. The source installation above avoids assuming Central publication. ## Test a tool and an approval gate without a provider From the extracted agent demo, build and run: ```sh "$PIPELOGIQ_SDK_DIR/mvnw" -q compile dependency:copy-dependencies java -cp 'target/classes:target/dependency/*' example.AgentHarnessExample ``` No API key, server or paid provider is needed. Expect `Success: true`, `Tools: [service_status]` and `Answer: The service is ready.` The [testing guide](/docs/java/testing) includes the complete harness and approval checks. A scripted planner exercises production handlers; it does not pretend to validate a real model's decisions. ## Connect a provider to an HTTP worker HTTP is the default in `pipelogiq.yml`; `transport: amqp` uses the same code. This example calls OpenAI when tasks arrive and can incur provider charges. Its read-only tool returns demonstration data rather than inspecting a production service. `src/main/java/example/AgentExample.java`: ```java package example; import io.pipelogiq.sdk.Pipelogiq; import io.pipelogiq.sdk.agent.AgentRuntime; import io.pipelogiq.sdk.agent.llm.Provider; import io.pipelogiq.sdk.agent.tools.*; import java.time.Duration; import java.util.Map; /** Production provider example. Provider/model/key are explicit application settings. */ public final class AgentExample { public static void main(String[] args) { if (args.length == 1 && args[0].equals("help")) { System.out.println("AgentExample starts a worker; set AGENT_MODEL and AGENT_API_KEY for OpenAI."); System.out.println("Submit separately with SubmitAgentExample, or use AgentHarnessExample offline."); return; } if (args.length != 0) throw new IllegalArgumentException("AgentExample accepts no arguments (or help)"); var runtime = AgentRuntime.builder() .react() .provider(Provider.OPENAI, required("AGENT_MODEL"), required("AGENT_API_KEY")) .configure(options -> { options.maxThinkSteps = 4; options.maxOutputTokens = 512; options.requestTimeout = Duration.ofSeconds(45); options.systemPrompt = "Use service_status to answer questions about the demo service. Be concise."; }) .tool( new ToolDefinition("service_status", "Read local service status").mutating(false), (parameters, context) -> NativeTool.Output.success(Map.of("status", "ready"))) .build(); Pipelogiq.worker() .config("pipelogiq.yml") .configure(worker -> runtime.registerHandlers(worker.registry())) .run(); } private static String required(String name) { String value = System.getenv(name); if (value == null || value.isBlank()) throw new IllegalArgumentException("Set " + name); return value; } } ``` This is configuration plus tool logic and runtime registration. No API client, transport switch, poller or shutdown hook is required in the worker entry point. The agent runtime registers its six built-in stage names before startup. Use the same `pipelogiq.yml` structure as [getting started](/docs/java/getting-started), and start the worker in Terminal A from the agent demo root: ```sh export PIPELOGIQ_API_URL='http://localhost:8081' export PIPELOGIQ_API_KEY='your-application-api-key' export AGENT_MODEL='your-available-model-id' export AGENT_API_KEY='your-provider-api-key' java -cp 'target/classes:target/dependency/*' example.AgentExample ``` Provider credentials stay in worker configuration. They are not sent as pipeline input. The process continues accepting tasks until Ctrl+C; starting it does not submit an agent request. ## Submit a typed agent request `src/main/java/example/SubmitAgentExample.java`: ```java package example; import io.pipelogiq.sdk.Pipelogiq; import io.pipelogiq.sdk.agent.AgentConstants; import io.pipelogiq.sdk.agent.model.OrchestratorInput; import io.pipelogiq.sdk.core.builders.StageOptions; import io.pipelogiq.sdk.core.client.StageRef; import java.time.Duration; /** Submit only while AgentExample is running. The worker may make billable provider calls. */ public final class SubmitAgentExample { public static void main(String[] args) { if (args.length == 0 || args[0].equals("help")) { System.out.println("SubmitAgentExample \"\""); return; } if (args.length != 3) throw new IllegalArgumentException("Expected "); var orchestrator = StageRef.of(AgentConstants.ORCHESTRATOR_HANDLER_NAME, OrchestratorInput.class); try (var client = Pipelogiq.connect("pipelogiq.yml")) { var created = client.pipeline("Agent request " + args[0]) .idempotencyKey("demo-agent-" + args[0]) .stage(orchestrator, new OrchestratorInput(args[2], args[1]), new StageOptions().maxRetries(0)) .send(); System.out.println("Pipeline ID: " + created.id()); var completed = client.pipelines().waitForCompletion(created.id(), Duration.ofMinutes(3)); System.out.println("Status: " + completed.status()); System.out.println("Answer: " + completed.context(AgentConstants.FINAL_RESPONSE, String.class)); } } } ``` The producer needs only the Pipelogiq application key. It references the runtime's orchestration contract through `StageRef`; subsequent stages are appended by the agent runtime. There is no hand-written JSON or status polling. In Terminal B, export the API URL/key and run: ```sh java -cp 'target/classes:target/dependency/*' example.SubmitAgentExample request-001 conversation-001 'Check the demo service status' ``` Expect a pipeline ID, terminal status and a model-generated answer. Keep the session ID to continue a conversation; use a new request ID for each new message. Repeating a request ID reuses the existing pipeline. A three-minute local observation timeout leaves server work running, so inspect its ID rather than accidentally resubmitting with a new key. Workers using the built-in agent names in the same application can consume the same work. Keep their tools, provider routing and stores compatible, or use separate applications for distinct agent configurations. The worker name is not an agent-routing key. ## Choose the reasoning mode and provider routes The default runtime mode is plan-and-execute: the planner produces an ordered tool plan. `.react()` enables repeated decisions after observing earlier results. Both use durable stages and the server's workflow state. The current adapters are `Provider.ANTHROPIC`, `OPENAI` and `OLLAMA`. Choose an explicit provider and model enabled for your account. Worker-owned `AgentOptions` holds credentials, base URLs and model routes. `options.routes` maps `Step.PLAN`, `THINK`, `SYNTHESIZE` and `CRITIC` to `Route(provider, model)`; `options.providers` supplies each provider's `Connection(apiKey, baseUrl)`. Per-run `RunOverrides` can carry mode/routing hints; they do not give model input permission to replace credentials or arbitrary endpoints. Anthropic supports image/document attachments, OpenAI images, and Ollama images. Transcribe audio into text first; unsupported attachment types are rejected before provider calls. ## Define tools with clear boundaries A native tool pairs a `ToolDefinition` with an implementation returning `NativeTool.Output`. Declare parameter schemas and `.mutating(true)` for tools that change external state. Tool code receives `StageContext`, including cooperative cancellation and authorized context values. Use `TypedNativeTool` to convert validated parameters into a record or POJO. This complete implementation can be registered with an appropriately declared `sku` parameter: ```java package example; import io.pipelogiq.sdk.agent.tools.TypedNativeTool; import io.pipelogiq.sdk.core.execution.StageContext; import java.util.Map; public final class StockTool extends TypedNativeTool { public record Input(String sku) {} public StockTool() { super(Input.class); } @Override protected Output executeTyped(Input input, StageContext context) { context.throwIfCancelled(); return Output.success(Map.of("sku", input.sku(), "available", 12)); } } ``` The returned quantity is demonstration data. Register it on an `AgentRuntime` builder with: ```java .tool(new ToolDefinition("stock", "Read demonstration stock") .mutating(false) .param("sku", new io.pipelogiq.sdk.agent.tools.ToolParam("body", "string", true)), new StockTool()) ``` For HTTP tools, configure a named `TargetApi` in worker options, then use `.http(method, path)`, `.target(name)` and parameter definitions. Authentication belongs to the configured target. `AuthHeader.contextBearer(key)` can read a credential provided by the authorized application in sensitive context. Parameter validation runs before `ToolPolicy`. The policy is your application authorization boundary; model tool selection and human confirmation do not replace it. Result references such as `{{ref:resultKey.path[0].value}}` can supply subsequent calls; missing references fail instead of silently becoming empty values. `AgentRuntime.builder().openApi(specUrlOrPath)` can import tool definitions. Advanced `OpenApiToolLoader` options filter operations and select a target name. Supported descriptions are JSON/YAML OpenAPI 3.x and Swagger 2 with local references. External references, multipart/form, non-object bodies and unsupported union schemas need explicit custom tools. Imported definitions do not supply target credentials or make all imported operations appropriate for your agent. ## Require approval for mutations Call `.confirmationRequired()` on the runtime builder, configure a `NotificationRouter` and submit a matching `ReplyTarget`. All three pieces matter: a flag alone does not deliver a usable approval request. Failed notification delivery does not authorize the mutation. A successful confirmation request leaves the stage waiting. An authorized approval service uses `client.stages().approve(stageId)` or `client.stages().reject(stageId, reason)`. Verify the reviewer and the actual waiting pipeline/stage. The approved batch retains its original parameters; rejected mutations remain unexecuted. The optional Telegram helpers connect notifications with inbound polling and validate pipeline/stage/chat/sender when handling approval callbacks. Their polling lifecycle is additional application work: start it after worker readiness and stop it before closing its borrowed API client. The source archive's `docs/advanced-runtime.md` includes the complete coordinated example; an ordinary agent without Telegram needs none of that setup. ## Use critics and budgets deliberately Critic modes are `OFF`, `FINAL`, `MUTATING` and `EVERY_STEP`. The default is `OFF`; `criticFailOpen` defaults to `true`. If review is a required gate, choose a mode and set `criticFailOpen=false`. `maxCriticRejections` bounds review/rethink loops. Think-step, token and cost settings bound runtime behavior but are not a provider billing cap. Per-run estimated-cost and input-token budgets are soft limits checked around calls, so one call can cross a limit. Configure prices for your actual model contract and use provider billing controls when you need spending enforcement. The example's output-token and request-timeout limits do not imply a fixed total cost. ## Persist conversations and memory Every newly constructed runtime has in-memory session and memory stores by default. A session lasts in that process, with a two-hour TTL from its last save; reads do not refresh it. The default history window is the newest 24 entries, not a token limit. Separate processes do not share these stores. Use `.withoutSessionHistory()` for explicit stateless runs. It disables cross-pipeline conversation history, not per-run context or memory. `runtime.sessions().clear(sessionId)` clears a conversation when sessions are enabled. Memory is written explicitly by the application or tools through `runtime.memory()`. Recall uses word matching, not vector embeddings; the runtime does not automatically extract durable memories from every conversation. ### Redis Add `com.pipelogiq:pipelogiq-sdk-redis:0.5.0` and install `sdk-redis -am` from source. An application startup method can own one shared pool for the worker lifetime: ```java import io.pipelogiq.sdk.redis.RedisAgentStores; import java.net.URI; try (var stores = RedisAgentStores.connect(URI.create(System.getenv("REDIS_URL")))) { var runtime = AgentRuntime.builder() .react() .provider(Provider.OPENAI, System.getenv("AGENT_MODEL"), System.getenv("AGENT_API_KEY")) .stores(stores) .tool(new ToolDefinition("service_status", "Read demo status").mutating(false), (parameters, context) -> NativeTool.Output.success(Map.of("status", "ready"))) .build(); Pipelogiq.worker().config("pipelogiq.yml") .configure(worker -> runtime.registerHandlers(worker.registry())).run(); } ``` This fragment uses the agent imports from `AgentExample`. `connect(uri, sessionTtl, memoryTtl, prefix)` customizes retention and namespace. `RedisAgentStores.using(existingClient)` borrows a caller-owned client; closing that pair does not close the supplied client. ### PostgreSQL Add `com.pipelogiq:pipelogiq-sdk-postgres:0.5.0` and install `sdk-postgres -am`. `new PostgresAgentStores(dataSource)` borrows your application's `DataSource`; `ensureSchema()` explicitly creates the required schema when that is part of your deployment policy. Managed migrations can provision equivalent tables instead. Attach stores with `.stores(stores)` during runtime construction or `stores.attachTo(runtime)` before startup. A JDBC URL/user/password constructor is also available. The stores close connections borrowed for operations, not a supplied data source. Stop workers before closing shared stores or pools. Both persistent backends replace whole session histories on save. Serialize concurrent conversations for the same session when overwriting updates would be unacceptable. Shared persistence does not add an automatic merge or distributed conversation lock. Continue with [agent tests](/docs/java/testing) and [lifecycle observations, traces and metrics](/docs/java/observability).