Build and test Java agents

The Java agent runtime turns planning, tool execution, confirmation, review and response synthesis into Pipelogiq stages. You can inspect each step through the same workflow model used by ordinary application handlers.

Start with a deterministic test, then connect a provider and worker. This separates your tool contracts and approval behavior from model quality and network availability.

Install the agent and testing modules

From your current Java SDK checkout:

cd "$PIPELOGIQ_SDK_DIR"
./mvnw -pl sdk-agent,sdk-testing -am install -DskipTests

Add these dependencies to the application from Get started with Java:

<dependency>
  <groupId>com.pipelogiq</groupId>
  <artifactId>pipelogiq-sdk-agent</artifactId>
  <version>0.5.0</version>
</dependency>
<dependency>
  <groupId>com.pipelogiq</groupId>
  <artifactId>pipelogiq-sdk-testing</artifactId>
  <version>0.5.0</version>
</dependency>

These instructions use local source installation while Maven Central publication is pending. The testing dependency is shown at application scope so this guide's standalone check can run from main; in an application test suite, use test scope instead.

Test a tool and an approval gate without a provider

Create src/main/java/example/AgentChecks.java:

package example;

import io.pipelogiq.sdk.agent.model.ToolCall;
import io.pipelogiq.sdk.agent.tools.NativeTool;
import io.pipelogiq.sdk.agent.tools.ToolDefinition;
import io.pipelogiq.sdk.testing.AgentTestHarness;
import java.util.List;
import java.util.Map;

public final class AgentChecks {
  private static AgentTestHarness approvalScenario() {
    return AgentTestHarness.create(options -> {
      options.useReActMode = false;
      options.requireConfirmationForMutations = true;
    }).withNativeTool(
        new ToolDefinition("reserve_demo_stock", "Simulate a stock reservation").mutating(true),
        (parameters, context) -> NativeTool.Output.success(Map.of("reserved", true)))
      .withScenario(scenario -> scenario
          .withPlan(new ToolCall("reserve_demo_stock", Map.of(), "reservation"))
          .withSynthesizedResponse("Demo reservation accepted"));
  }

  private static void check(boolean condition, String message) {
    if (!condition) throw new AssertionError(message);
  }

  public static void main(String[] args) {
    var read = AgentTestHarness.create()
        .withNativeTool(
            new ToolDefinition("service_status", "Read simulated service status").mutating(false),
            (parameters, context) -> NativeTool.Output.success(Map.of("status", "ready")))
        .withScenario(scenario -> scenario
            .thenCallTool("service_status").thenFinishWith("Service is ready"))
        .run("Check the service", "test-session");
    check(read.isSuccess(), read.errorMessage());
    check(read.toolCallsExecuted().equals(List.of("service_status")), "Expected one read tool");
    check("Service is ready".equals(read.finalResponse()), "Unexpected final response");

    var waiting = approvalScenario().run("Reserve demo stock");
    check(waiting.isWaitingForApproval(), "Expected a waiting approval");
    check(waiting.toolCallsExecuted().isEmpty(), "Mutation must wait for approval");

    var approved = approvalScenario().withApprovalDecision(true).run("Reserve demo stock");
    check(approved.isSuccess(), approved.errorMessage());
    check(approved.toolCallsExecuted().equals(List.of("reserve_demo_stock")),
        "Expected the approved mutation");

    var rejected = approvalScenario().withApprovalDecision(false, "Customer cancelled")
        .run("Reserve demo stock");
    check(rejected.toolCallsExecuted().isEmpty(), "Rejected mutation must not execute");
    System.out.println("Agent checks passed: read, waiting, approved, rejected");
  }
}

Rebuild the application and run:

"$PIPELOGIQ_SDK_DIR/mvnw" -q compile dependency:copy-dependencies
java -cp 'target/classes:target/dependency/*' example.AgentChecks

This check needs no Pipelogiq server, provider key or live inventory system. The planner and business tools are simulated; the harness executes the production agent handlers with JSON serialization, context patches and appended stages. It exposes stage results, executed tool names, notifications, context and final response for assertions.

The harness does not simulate server retries, leases, transport failures, durable transactions or concurrent scheduling. Use server integration tests for those. It also does not expose external HTTP tool registration; use native fixtures here and protocol fixtures for HTTP adapters.

Connect a provider to an HTTP worker

The following example makes real provider calls, which can incur charges. Its service_status tool returns explicit demonstration data; it does not probe a production service.

Create src/main/java/example/ServiceAgent.java. It uses OrdersApp from the getting-started project:

package example;

import io.pipelogiq.sdk.agent.AgentOptions;
import io.pipelogiq.sdk.agent.AgentPipeline;
import io.pipelogiq.sdk.agent.AgentRuntime;
import io.pipelogiq.sdk.agent.llm.Provider;
import io.pipelogiq.sdk.agent.model.OrchestratorInput;
import io.pipelogiq.sdk.agent.store.InMemorySessionStore;
import io.pipelogiq.sdk.agent.tools.NativeTool;
import io.pipelogiq.sdk.agent.tools.ToolDefinition;
import io.pipelogiq.sdk.agent.tools.ToolRegistry;
import io.pipelogiq.sdk.core.api.PipelogiqApiClient;
import io.pipelogiq.sdk.http.HttpPipelineRunner;
import java.time.Duration;
import java.util.Map;
import java.util.UUID;

public final class ServiceAgent {
  private static String required(String name) {
    String value = System.getenv(name);
    if (value == null || value.isBlank()) throw new IllegalArgumentException("Set " + name);
    return value;
  }

  public static void main(String[] args) throws Exception {
    var options = OrdersApp.options("service-agent-java");
    var agent = new AgentOptions().react();
    agent.provider = Provider.OPENAI;
    agent.model = required("AGENT_MODEL");
    agent.apiKey = required("AGENT_API_KEY");
    agent.maxThinkSteps = 8;
    agent.maxInputTokensPerRun = 50000;
    agent.systemPrompt = "Report the demonstrated service status. Do not claim to inspect a real service.";

    var tools = new ToolRegistry().register(
        new ToolDefinition("service_status", "Read demonstration service status").mutating(false),
        (parameters, context) -> {
          context.throwIfCancelled();
          return NativeTool.Output.success(Map.of("service", "orders-demo", "status", "ready"));
        });
    var runtime = new AgentRuntime(agent, tools).sessions(new InMemorySessionStore());

    try (var worker = new HttpPipelineRunner(options);
         var api = new PipelogiqApiClient(options)) {
      runtime.registerHandlers(worker.registry());
      Runtime.getRuntime().addShutdownHook(new Thread(worker::close));
      var lifetime = worker.start();
      worker.awaitReady(Duration.ofSeconds(30));
      try (var pipeline = AgentPipeline.create(
          new OrchestratorInput("Check the demonstration service", "demo-session"), api)
          .withIdempotencyKey("service-agent:" + UUID.randomUUID())) {
        long id = pipeline.send().path("id").asLong();
        System.out.println("Agent pipeline ID: " + id);
        System.out.println(OrdersApp.waitFor(api, id).toPrettyString());
      }
      worker.stop(Duration.ofSeconds(30));
      lifetime.join();
    }
  }
}

Set the same Pipelogiq variables as before, plus a provider model enabled for your account and its key:

export AGENT_MODEL='your-enabled-model-id'
export AGENT_API_KEY='your-provider-api-key'
java -cp 'target/classes:target/dependency/*' example.ServiceAgent

The code stores the final answer under agent:finalResponse in pipeline context. It can be retrieved through the API even without a notification channel. The sample client waits up to 60 seconds; longer provider runs can exceed that observation window. For a long-running deployment, run the worker continuously as described in worker operations and submit from a separate client.

Each runtime registers six shared handler names. Workers in the same application that register those names compete for the same agent work. Configure them consistently, or isolate distinct agent configurations into separate applications; a worker name is not an agent-routing key.

Choose the reasoning mode and provider routes

The default mode is plan-and-execute: the planner produces an ordered set of tool calls. .react() enables a loop in which the model chooses the next action after observing previous results. Both persist progress through stages; ReAct appends subsequent stages with the current stage result.

Production adapters support Anthropic Messages, OpenAI Chat Completions and Ollama Chat, including provider-native tool calls. Set model names explicitly. Configure credentials and base URLs in worker-owned AgentOptions, never in model-generated input.

AgentOptions.routes maps Step.PLAN, THINK, SYNTHESIZE and CRITIC to Route(provider, model). The providers map holds each provider's Connection(apiKey, baseUrl). RunOverrides supplies per-run mode/routing hints, not credentials or arbitrary endpoint access. For Ollama, configure the worker to reach the Ollama host and select a model already available there.

Attachments are provider-specific: Anthropic supports images/documents, OpenAI images, and Ollama images. Transcribe audio before supplying it as text. Unsupported attachment types are rejected before provider requests.

Define tools with clear boundaries

A native tool has a ToolDefinition and an implementation returning NativeTool.Output. Mark mutations explicitly with .mutating(true) and provide meaningful descriptions and parameter schemas. Native tools receive StageContext, so they can honor cancellation, read authorized context and produce controlled context updates.

For HTTP tools, configure a named TargetApi in AgentOptions.targetApis, then register a tool with .http(method, path), .target(name) and ToolParam definitions. Authentication belongs to the target or worker configuration. AuthHeader.contextBearer(key) can read a sensitive credential supplied by the authorized application.

Parameters are validated before ToolPolicy. Result references such as {{ref:resultKey.path[0].value}} can feed later calls; missing references fail instead of silently becoming an empty value. Policy is an application authorization boundary, separate from the model's selection of a tool and a user's confirmation.

OpenApiToolLoader imports JSON/YAML OpenAPI 3.x and Swagger 2 descriptions, including local refs, operation filters and parameter schemas. External refs, oneOf/anyOf, multipart/form and non-object request bodies need explicitly defined custom tools. Review generated tool mutation flags and scope before exposing an imported API to an agent.

Require approval for mutations

Call confirmationRequired() before constructing the runtime. Configure a NotificationRouter with a real channel implementation and supply OrchestratorInput.replyTo with the intended ReplyTarget. A successful confirmation notification moves the stage into WaitingForApproval; a delivery failure does not authorize the mutation.

Resume through PipelogiqApiClient.resumeStage(stageId, true, null) or reject with resumeStage(stageId, false, reason). Your approval surface must authenticate the reviewer and verify that the decision belongs to the expected pipeline, waiting stage and user. Telegram adapters implement their own pipeline/stage/chat/user callback checks. A notification reaching a channel is not itself an approval.

The approved batch uses its original tool parameters. A final approved mutation batch proceeds to response synthesis. Rejected mutations remain unexecuted; the agent can still return a response describing the rejection. Exercise both outcomes with the deterministic harness before connecting a live mutation tool.

Use critics and budgets deliberately

Critic modes are OFF, FINAL, MUTATING and EVERY_STEP. The default is OFF, and criticFailOpen defaults to true. If critic review is a required gate, enable the relevant mode and set criticFailOpen=false. Rejection can send feedback back to the thinker; maxCriticRejections bounds that loop.

Input-token and estimated-cost budgets are per-run soft limits. They are checked around calls and completion, so one provider call may cross a limit. Model prices are estimates, not a live billing feed. Set worker-owned prices for your actual model contract, and use provider billing controls when you need spending enforcement.

Persist conversations and memory

An AgentSessionStore retains conversation history across pipeline runs sharing a sessionId. An in-memory store lasts only as long as its owning process; the default session TTL is two hours from the last save, and reads do not refresh it. The one-shot ServiceAgent example demonstrates wiring but exits after one run, so its store does not survive process restarts.

For shared persistence, install pipelogiq-sdk-redis or pipelogiq-sdk-postgres. Inject their session/memory stores into the runtime. Redis constructors can own a client created from a URI, or accept a caller-owned RedisClient. PostgreSQL stores can use a caller-owned DataSource; call ensureSchema() when your deployment intentionally allows the stores to create their tables, or provision that schema through your migrations.

Memory storage is explicit: application code or a native tool stores entries, and the runtime adds recalled entries to prompts. Recall ranks word matches; it is not embedding or vector search. Use separate namespaces and deliberate retention policies for unrelated applications. Store operations are synchronous, and connection/pool tuning belongs to the injected client or data source.

Continue to the Java reference for package mapping, defaults and interoperability limits.