Agentic Communication Patterns

Patterns for enabling effective communication and collaboration between agents

Tool Calling / Function Calling

Description

Want your AI agents to do more than just chat? Tool calling enables LLM agents to execute functions, from simple calculations to complex API calls. The Model Context Protocol (MCP) is a lightweight, flexible bridge that makes this possible, simplifying tool integration and letting your agents interact with APIs, databases, or scripts without complex setup. With just a few lines of code, your AI can access live data or trigger real-world actions.

Key Communication Protocols

  • Function Registration

    • Tools are registered with metadata describing their purpose, parameters, and return types.
    • The @mcp.tool() decorator makes functions agent-compatible with structured input validation.
    • Frameworks provide standardized ways to define and expose tool capabilities.

  • Structured Invocation

    • Agents generate structured requests (typically JSON) specifying the tool and arguments.
    • Includes parameter validation and type checking before execution.
    • Pydantic ensures structured input validation for robust tool interactions.

  • Execution & Response

    • Tools execute in a controlled environment with proper error handling.
    • Results are returned in a structured format for agent interpretation.
    • MCP servers can run locally or be exposed via ngrok for remote access.

  • Security & Validation

    • Tools run with appropriate permissions and resource limits.
    • Input validation and sanitization before execution.
    • Authentication, authorization, and sandboxing for secure tool execution.

Step-by-Step: Building a Currency Converter Tool

1. Set Up Your Environment

First, create a Python virtual environment and install dependencies:

pip install fastmcp pydantic crewai python-dotenv

2. Define Your MCP Tool

Create a Python file (e.g., currency_tool.py) for your currency converter. This example uses a mock conversion rate, but you can swap in a real API for live data:

@mcp.tool()
async def convert_currency(request: CurrencyConversionInput) -> str:

The @mcp.tool() decorator makes this function agent-compatible, and pydantic ensures structured input validation.

3. Run the MCP Server

Start your server locally:

python currency_tool.py

This launches an MCP server at http://localhost:8000. Your tool is now ready to accept requests!

4. Expose Your Server with ngrok

To let agents access your tool, we need to expose your local server using ngrok:

  • Install ngrok and authenticate it.
  • Run: ngrok http 8000
  • Copy the generated URL (e.g., https://abc123.ngrok-free.app).

Tip: ngrok's free tier is great for testing, but for production, consider a paid plan or deploy your server to a cloud provider.

5. Connect to CrewAI

Integrate your tool with a CrewAI agent:

server_config = "https://abc123.ngrok-free.app"  # use ngrok url

with MCPServerAdapter(server_config) as currency_tools:
    currency_agent = Agent(
        role="Currency Exchange Expert",
        goal="Help with currency conversions and exchange rate information",
        backstory="You're a knowledgeable financial assistant with access to real-time currency conversion tools.",
        tools=currency_tools,  # <- Pass the list of actual tool instances
        verbose=False
    )

Your agent can now call convert_currency seamlessly!

stdio vs. SSE: Which to Use?

MCP supports two communication modes for your server:

stdio

Ideal for local development or CLI-based tools. It uses standard input/output, making it simple and fast for single-machine setups.

SSE (Server-Sent Events)

Best for remote servers or real-time applications. SSE streams data over HTTP, perfect for progressive responses or cloud-hosted tools.

Pro Tip: Use stdio for quick prototyping and SSE for production-grade, internet-facing servers.

Securing Your MCP Implementation

To ensure your MCP server is secure, adopt these best practices:

  • Authentication & Authorization: Use robust mechanisms to verify users and restrict access.
  • Session Management: Use strong, non-deterministic session IDs and consider binding them to user-specific information.
  • Sandboxing: Isolate MCP servers (local or remote) to limit exposure to untrusted content.
  • Input Validation: Validate prompts and verify tool integrity to prevent malicious inputs.
  • Least Privilege: Grant minimal permissions to MCP components and agents.
  • Monitoring & Logging: Implement systems to detect and log suspicious activities.
  • User Confirmation: Require explicit user approval for sensitive actions triggered by the agent.

Next Steps

  • Enhance Your Tool: Replace the mock rates with a real API (e.g., exchangerate-api.com).
  • Explore More Tools: Add more MCP tools for tasks like stock lookups or weather updates.

RAG (Retrieval Augmented Generation)

Description

RAG (Retrieval Augmented Generation) is a pattern that enhances LLM responses by incorporating relevant information from external knowledge sources. This pattern is particularly useful for grounding LLM outputs in factual data and providing up-to-date information. RAG combines the power of information retrieval with generative AI to produce more accurate and contextually relevant responses.

Implementation Example 

# Pseudocode for RAG (Retrieval Augmented Generation) Pipeline

# 1. Store documents with embeddings in a vector store
vector_store = VectorStore()
for doc in documents:
    embedding = embed(doc)
    vector_store.add(doc, embedding)

# 2. When a query comes in:
def process_query(query):
    # a. Embed the query
    query_embedding = embed(query)
    # b. Retrieve top-k similar documents
    relevant_docs = vector_store.search(query_embedding, top_k)
    # c. Concatenate retrieved docs as context
    context = join([doc.content for doc in relevant_docs])
    # d. Pass query and context to LLM
generated_response = llm_generate(query, context)
    # e. Return the response
    return generated_response

# Example usage
response = process_query("What is RAG?")
print(response)

Agent-to-Agent Protocol (A2A)

Description

Google's A2A is an open protocol designed for horizontal integration – secure and standardized communication between diverse AI agents across different platforms and vendors. This protocol allows agents from different platforms and vendors to discover each other's capabilities, exchange messages, and collaborate on tasks. It uses signed messages and capability discovery to ensure secure cross-platform interactions, making it ideal for distributed agent systems and cloud-based AI applications. Note that this is different from local multi-agent frameworks like CrewAI and AutoGen, which focus on coordinating agents within a single system.

Key Communication Protocols

  • Capability Discovery (Agent Cards)

    • Agents publish public JSON "Agent Cards" (e.g., at /.well-known/agent.json) describing their functionalities, endpoints, security, and data formats.
    • Client agents dynamically discover and select suitable remote agents based on these cards.

  • Task Delegation

    • A client agent sends a "Task" (a specific work unit with a unique ID) to a remote agent.
    • Tasks include metadata and a "message" containing details (text, files, etc.).

  • Structured Messaging & Artifacts

    • Communication uses signed, structured messages based on a shared schema.
    • Results are returned as "Artifacts," which can be multi-part (e.g., text, files), crucial for rich data exchange.

  • Asynchronous Operations & Notifications

    • Supports long-running tasks, with clients receiving updates via push notifications or Server-Sent Events (SSE).

Conceptual Example: A2A Interaction 

# Pseudocode for A2A (Agent-to-Agent) Interaction

# 1. Remote agent publishes its capabilities (agent card)
remote_agent_card = {
    "id": "data-processor-001",
    "capabilities": ["process_data", "generate_report"],
    "api_url": "https://api.example.com/agents/data-processor-001"
}

# 2. Client agent discovers remote agent
client_agent.discover(remote_agent_card)

# 3. Client agent delegates a task to the remote agent
task = {
    "id": generate_unique_id(),
    "capability": "process_data",
    "message": {
        "data_type": "sales_report",
        "format": "json",
        "parameters": {"region": "EMEA", "period": "Q3"}
    },
    "metadata": {"sender": client_agent.id, "timestamp": now()}
}

result = remote_agent.process_task(task)

# 4. Client agent receives the result
print(result)

Agent Communication Protocol (ACP)

Description

Agent Communication Protocol (ACP) provides a structured framework for agent interactions, focusing on local-first, REST-native communication. It enables secure and efficient dialogue between agents through standardized message formats, event-driven architecture, and metadata management. This protocol is particularly useful for privacy-sensitive scenarios and edge computing applications.

Key Communication Protocols

  • REST-native Messaging

    • Standard HTTP methods for agent communication.
    • Structured request/response formats for clear interaction patterns.

  • Event-Driven Architecture

    • Asynchronous communication through event publishing and subscription.
    • Support for real-time updates and notifications.

  • Structured Dialogue

    • Standardized message formats for different types of interactions.
    • Support for multi-turn conversations and context management.

  • Security & Authentication

    • Built-in support for authentication and authorization.
    • Secure message exchange and data protection.

  • Message Performative Semantics (FIPA-inspired)

    • Messages include a performative field to indicate intent, such as: inform, request, agree, refuse, propose, inform-result.
    • This helps agents interpret the purpose of a message and respond appropriately.
    • Example: message_type="request" # message.performative = "request"

Conceptual Example: ACP Communication 

# Pseudocode for ACP (Agent Communication Protocol) Communication

class ACPAgent:
    def __init__(self, id):
        self.id = id
        self.handlers = {}
    def register_handler(self, message_type, handler):
        self.handlers[message_type] = handler
    def receive_message(self, message):
        # Auto-dispatch based on performative / type
        handler = self.handlers.get(getattr(message, 'performative', None) or message.message_type)
        if handler:
            handler(message)

# 1. Define two agents with unique IDs
agent1 = ACPAgent("agent1")
agent2 = ACPAgent("agent2")

# 2. Register a handler for request messages on agent1
def handle_request(message):
    # Process the request and prepare a response
    response = ACPMessage(
        message_type="response",
        performative="inform-result",  # FIPA-inspired performative
        content={"status": "success", "data": "Request processed"},
        sender_id=agent1.id,
        recipient_id=message.sender_id
    )
    agent1.send_message(response)

agent1.register_handler("request", handle_request)

# 3. agent2 sends a request message to agent1
request = ACPMessage(
    message_type="request",
    performative="request",  # FIPA-inspired performative
    content={"action": "process_data", "data": "sample data"},
    sender_id=agent2.id,
    recipient_id=agent1.id
)
agent2.send_message(request)

# 4. agent1 receives and handles the request
agent1.receive_message(request)