CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/maven-dev-langchain4j--langchain4j-agentic

LangChain4j Agentic Framework provides a comprehensive Java library for building multi-agent AI systems with support for workflow orchestration, supervisor agents, planning-based execution, declarative configuration, agent-to-agent communication, and human-in-the-loop workflows.

Overview
Eval results
Files

a2a.mddocs/advanced/

Agent-to-Agent Communication (A2A)

Distributed agent communication enabling agents to invoke and interact with agents running in remote services.

Overview

A2A enables:

  • Distributed agent systems across services
  • Microservices coordination
  • Remote agent invocation
  • Service discovery patterns
  • Load balancing and failover

Factory Methods

static A2AClientBuilder<UntypedAgent> a2aBuilder(String serverUrl);
static <T> A2AClientBuilder<T> a2aBuilder(String serverUrl, Class<T> agentServiceClass);

Quick Start:

UntypedAgent remoteAgent = AgenticServices.a2aBuilder("https://agent-service.example.com/api/v1")
    .name("remote-analysis-agent")
    .description("Remote agent for data analysis")
    .build();

Object result = remoteAgent.invoke("Analyze this dataset");

Configuration

A2AClientBuilder

interface A2AClientBuilder<T> {
    T build();
    A2AClientBuilder<T> name(String name);
    A2AClientBuilder<T> description(String description);
    A2AClientBuilder<T> outputKey(String outputKey);
    A2AClientBuilder<T> outputKey(Class<? extends TypedKey<?>> outputKey);
    A2AClientBuilder<T> output(Function<AgenticScope, Object> output);
    A2AClientBuilder<T> errorHandler(Function<ErrorContext, ErrorRecoveryResult> errorHandler);
    A2AClientBuilder<T> listener(AgentListener listener);
}

Example:

UntypedAgent a2aAgent = AgenticServices.a2aBuilder("https://remote.example.com/api")
    .name("remote-processor")
    .description("Processes data on remote service")
    .outputKey("processed_data")
    .errorHandler(errorContext -> {
        System.err.println("A2A error: " + errorContext.error().getMessage());
        return new ErrorRecoveryResult("fallback", true);
    })
    .listener(myListener)
    .build();

Common Patterns

Distributed Pipeline

interface DistributedPipeline {
    @SequenceAgent(
        name = "distributed-etl",
        subAgents = {
            LocalExtractor.class,
            RemoteTransformer.class,
            RemoteValidator.class,
            LocalLoader.class
        }
    )
    String runETL(String source);
}

interface RemoteTransformer {
    @A2AClientAgent(
        name = "transformer",
        serverUrl = "https://transform-service.example.com/api",
        outputKey = "transformed_data"
    )
    String transform(AgenticScope scope) {
        return (String) scope.readState("raw_data");
    }
}

Microservices Coordination

UntypedAgent coordinator = AgenticServices.supervisorBuilder()
    .chatModel(chatModel)
    .subAgents(
        // Local agents
        localAuthAgent,
        localRoutingAgent,

        // Remote microservices
        AgenticServices.a2aBuilder("https://inventory.example.com/api")
            .name("inventory-service")
            .build(),

        AgenticServices.a2aBuilder("https://payment.example.com/api")
            .name("payment-service")
            .build(),

        AgenticServices.a2aBuilder("https://shipping.example.com/api")
            .name("shipping-service")
            .build()
    )
    .maxAgentsInvocations(15)
    .build();

Object result = coordinator.invoke("Process order #12345");

Hybrid Local-Remote Processing

UntypedAgent hybridWorkflow = AgenticServices.parallelBuilder()
    .subAgents(
        // Local agents
        localSentimentAnalyzer,
        localKeywordExtractor,

        // Remote specialized services
        AgenticServices.a2aBuilder("https://translation.example.com/api")
            .name("translator")
            .outputKey("translated_text")
            .build(),

        AgenticServices.a2aBuilder("https://summarization.example.com/api")
            .name("summarizer")
            .outputKey("summary")
            .build(),

        AgenticServices.a2aBuilder("https://entity-recognition.example.com/api")
            .name("ner")
            .outputKey("entities")
            .build()
    )
    .output(scope -> Map.of(
        "sentiment", scope.readState("sentiment"),
        "keywords", scope.readState("keywords"),
        "translation", scope.readState("translated_text"),
        "summary", scope.readState("summary"),
        "entities", scope.readState("entities")
    ))
    .build();

Error Handling

Retry on Network Errors

UntypedAgent resilientA2A = AgenticServices.a2aBuilder("https://remote.example.com/api")
    .name("resilient-remote-agent")
    .errorHandler(errorContext -> {
        Throwable error = errorContext.error();
        AgenticScope scope = errorContext.agenticScope();

        int retries = scope.readState("retry_count", 0);

        if (error instanceof NetworkException && retries < 3) {
            scope.writeState("retry_count", retries + 1);
            System.out.println("Retrying... (attempt " + (retries + 1) + ")");
            return new ErrorRecoveryResult(null, true);
        }

        System.err.println("Remote agent failed after retries");
        return new ErrorRecoveryResult(null, false);
    })
    .build();

Fallback to Local Processing

UntypedAgent fallbackA2A = AgenticServices.a2aBuilder("https://remote.example.com/api")
    .name("remote-with-fallback")
    .errorHandler(errorContext -> {
        System.err.println("Remote unavailable, using local fallback");

        String input = (String) errorContext.agenticScope().readState("input");
        String localResult = localFallbackAgent.invoke(input).toString();

        return new ErrorRecoveryResult(localResult, false);
    })
    .build();

Circuit Breaker

class CircuitBreakerErrorHandler implements Function<ErrorContext, ErrorRecoveryResult> {
    private int failureCount = 0;
    private boolean circuitOpen = false;
    private static final int THRESHOLD = 5;

    @Override
    public ErrorRecoveryResult apply(ErrorContext context) {
        if (circuitOpen) {
            System.err.println("Circuit breaker open");
            return new ErrorRecoveryResult("Service unavailable", false);
        }

        failureCount++;

        if (failureCount >= THRESHOLD) {
            circuitOpen = true;
            System.err.println("Circuit breaker opened after " + failureCount + " failures");
        }

        return new ErrorRecoveryResult(null, false);
    }

    public void reset() {
        failureCount = 0;
        circuitOpen = false;
    }
}

UntypedAgent protectedA2A = AgenticServices.a2aBuilder("https://remote.example.com/api")
    .name("circuit-breaker-agent")
    .errorHandler(new CircuitBreakerErrorHandler())
    .build();

Monitoring

class A2AMonitoringListener implements AgentListener {
    @Override
    public void beforeAgentInvocation(AgentRequest request) {
        if (isA2AAgent(request.agentType())) {
            System.out.println("A2A Request: " + request.agentName());
            System.out.println("Arguments: " + request.arguments());
        }
    }

    @Override
    public void afterAgentInvocation(AgentResponse response) {
        if (isA2AAgent(response.agentType())) {
            System.out.println("A2A Response: " + response.agentName());
            System.out.println("Duration: " + response.duration().toMillis() + "ms");
        }
    }

    @Override
    public void onAgentInvocationError(AgentInvocationError error) {
        if (isA2AAgent(error.agentType())) {
            System.err.println("A2A Error: " + error.agentName());
            System.err.println("Error: " + error.error().getMessage());
        }
    }

    private boolean isA2AAgent(Class<?> agentType) {
        return agentType.getName().contains("A2AClient");
    }
}

UntypedAgent monitoredWorkflow = AgenticServices.sequenceBuilder()
    .listener(new A2AMonitoringListener())
    .subAgents(
        localAgent,
        AgenticServices.a2aBuilder("https://remote.example.com/api")
            .name("remote-agent")
            .build()
    )
    .build();

Service Discovery

class ServiceDiscovery {
    private final Map<String, String> registry = new ConcurrentHashMap<>();

    public void register(String serviceName, String serviceUrl) {
        registry.put(serviceName, serviceUrl);
    }

    public String discover(String serviceName) {
        return registry.get(serviceName);
    }
}

// Use with A2A
ServiceDiscovery discovery = new ServiceDiscovery();
discovery.register("analytics", "https://analytics.example.com/api");
discovery.register("ml", "https://ml.example.com/api");

UntypedAgent dynamicWorkflow = AgenticServices.sequenceBuilder()
    .subAgents(
        localAgent,

        AgenticServices.a2aBuilder(discovery.discover("analytics"))
            .name("analytics-service")
            .build(),

        AgenticServices.a2aBuilder(discovery.discover("ml"))
            .name("ml-service")
            .build()
    )
    .build();

Load Balancing

class LoadBalancedA2AAgent implements UntypedAgent {
    private final List<UntypedAgent> agents;
    private final AtomicInteger counter = new AtomicInteger(0);

    public LoadBalancedA2AAgent(List<String> serviceUrls) {
        this.agents = serviceUrls.stream()
            .map(url -> AgenticServices.a2aBuilder(url)
                .name("load-balanced-agent")
                .build())
            .collect(Collectors.toList());
    }

    @Override
    public Object invoke(String input) {
        // Round-robin load balancing
        int index = counter.getAndIncrement() % agents.size();
        return agents.get(index).invoke(input);
    }

    @Override
    public Object invoke(Object memoryId, String input) {
        int index = counter.getAndIncrement() % agents.size();
        return agents.get(index).invoke(memoryId, input);
    }

    // Implement other methods...
}

UntypedAgent loadBalanced = new LoadBalancedA2AAgent(List.of(
    "https://service1.example.com/api",
    "https://service2.example.com/api",
    "https://service3.example.com/api"
));

Security

Authentication

class AuthenticatedA2ABuilder {
    public static UntypedAgent build(String url, String apiKey) {
        return AgenticServices.a2aBuilder(url)
            .name("authenticated-agent")
            .listener(new AgentListener() {
                @Override
                public void beforeAgentInvocation(AgentRequest request) {
                    request.agenticScope().writeState("auth_header", "Bearer " + apiKey);
                }
            })
            .build();
    }
}

UntypedAgent secureAgent = AuthenticatedA2ABuilder.build(
    "https://secure-service.example.com/api",
    System.getenv("API_KEY")
);

Mutual TLS

UntypedAgent mtlsAgent = AgenticServices.a2aBuilder("https://secure.example.com/api")
    .name("mtls-agent")
    .listener(new AgentListener() {
        @Override
        public void beforeAgentInvocation(AgentRequest request) {
            request.agenticScope().writeState("client_cert", loadClientCertificate());
        }
    })
    .build();

Testing

Mock A2A Agent

class MockA2AAgent implements UntypedAgent {
    private final String mockResponse;

    public MockA2AAgent(String mockResponse) {
        this.mockResponse = mockResponse;
    }

    @Override
    public Object invoke(String input) {
        return mockResponse;
    }

    @Override
    public Object invoke(Object memoryId, String input) {
        return mockResponse;
    }

    // Implement other methods...
}

// Use in tests
@Test
void testDistributedWorkflow() {
    UntypedAgent mockRemote = new MockA2AAgent("mock analysis result");

    UntypedAgent testWorkflow = AgenticServices.sequenceBuilder()
        .subAgents(localAgent, mockRemote)
        .build();

    Object result = testWorkflow.invoke("test input");

    assertEquals("expected output", result);
}

Declarative API

Example:

interface DistributedSystem {
    @SequenceAgent(
        name = "distributed-workflow",
        subAgents = {LocalPreprocessor.class, RemoteAnalyzer.class, LocalReporter.class}
    )
    String processData(String input);
}

interface RemoteAnalyzer {
    @A2AClientAgent(
        name = "remote-analyzer",
        serverUrl = "https://analytics.example.com/api/v1",
        outputKey = "analysis_result"
    )
    String analyze(AgenticScope scope) {
        String data = (String) scope.readState("preprocessed_data");
        return data;
    }
}

DistributedSystem system = AgenticServices.createAgenticSystem(
    DistributedSystem.class,
    chatModel
);

String result = system.processData("sensor_data.csv");

Best Practices

  1. Implement timeouts - Don't wait forever for remote services
  2. Use circuit breakers - Protect against cascading failures
  3. Implement retries - Handle transient network errors
  4. Monitor latency - Track remote call performance
  5. Secure communications - Use authentication and encryption
  6. Handle failures gracefully - Provide fallbacks
  7. Test with mocks - Unit test without actual remote calls
  8. Document endpoints - Maintain clear service contracts

See Also

  • Error Handling - Comprehensive error handling
  • Sequential Workflows - Integrating A2A in sequences
  • Supervisor Agents - Coordinating remote agents
  • Declarative Annotations - @A2AClientAgent annotation

Install with Tessl CLI

npx tessl i tessl/maven-dev-langchain4j--langchain4j-agentic

docs

advanced

a2a.md

error-handling.md

human-in-loop.md

persistence.md

index.md

tile.json