agent-protocol

name	agent-protocol
description	Design and implement AI agent communication protocols including MCP tool schemas, Google A2A protocol, OpenAI function calling, structured inter-agent messaging, and protocol negotiation. Use when building multi-agent systems, defining tool interfaces, implementing agent-to-agent communication, or standardizing LLM tool calling patterns across platforms.
license	MIT + Commons Clause
metadata	{"version":"1.0.0","author":"borghei","category":"engineering","domain":"ai-agent-systems","tier":"POWERFUL","updated":"2026-03-09T00:00:00.000Z","frameworks":"mcp, a2a, openai-functions, langchain-tools"}

Agent Protocol

The agent designs tool schemas for MCP, Google A2A, and OpenAI Function Calling protocols. It implements transport selection, capability discovery, authentication flows (OAuth 2.1, API keys), structured error handling, rate limiting, and protocol bridges for heterogeneous agent ecosystems.

Core Capabilities

1. Protocol Selection and Comparison

• MCP (Model Context Protocol): Anthropic's standard for tool/resource/prompt serving
• Google A2A (Agent-to-Agent): Agent card discovery, task lifecycle, streaming
• OpenAI Function Calling: JSON Schema tool definitions, parallel calls, strict mode
• LangChain/LangGraph Tools: Python-native tool wrappers with callback integration
• Custom Protocols: WebSocket, gRPC, and event-driven agent messaging

2. Tool Schema Design

• JSON Schema validation for inputs and outputs
• Semantic naming conventions that improve agent tool selection
• Description engineering for maximum LLM comprehension
• Required vs optional parameter design
• Enum constraints and default value strategies

3. Transport and Discovery

• stdio, SSE, and WebSocket transport for MCP
• HTTP+JSON-RPC for A2A task management
• Agent card and capability advertisement
• Health checking and graceful degradation
• Protocol version negotiation

4. Security and Authentication

• OAuth 2.1 flows for MCP remote servers
• API key rotation and scoping
• Request signing and verification
• Rate limiting per agent identity
• Audit logging for all inter-agent calls

When to Use

• Designing tool interfaces for LLM-powered agents
• Building MCP servers that expose APIs to Claude, Cursor, or other clients
• Implementing agent-to-agent communication in multi-agent systems
• Bridging between different agent protocols (MCP to A2A, etc.)
• Standardizing tool calling patterns across a team or organization
• Debugging agent tool selection failures

Protocol Comparison Matrix

Feature	MCP	A2A	OpenAI Functions	LangChain Tools
Transport	stdio/SSE/WebSocket	HTTP+JSON-RPC	HTTP REST	In-process
Discovery	Server capabilities	Agent cards	API spec	Registry
Streaming	SSE notifications	SSE streaming	Streaming deltas	Callbacks
Auth	OAuth 2.1	Agent auth	API key	N/A
State	Resources + context	Task lifecycle	Conversation	Memory
Multi-turn	Sampling	Task updates	Thread context	Chain state
File handling	Resource URIs	Artifact parts	File search	Document loaders
Best for	Tool serving	Agent networks	Single-model tools	Python pipelines

Decision Framework

What are you building?
│
├─ Tools for a single LLM client (Claude, Cursor, Copilot)
│  └─ Use MCP — it's the native protocol for tool serving
│
├─ Agent-to-agent communication across organizations
│  └─ Use A2A — designed for cross-boundary agent discovery and delegation
│
├─ Tools for OpenAI models specifically
│  └─ Use OpenAI Function Calling — tightest integration
│
├─ Python pipeline with multiple chained tools
│  └─ Use LangChain Tools — simplest for in-process orchestration
│
└─ Heterogeneous agent ecosystem (multiple protocols)
   └─ Use Protocol Bridge pattern — translate between protocols at boundaries

MCP Tool Schema Design

Anatomy of a Well-Designed Tool

{
  "name": "search_documents",
  "description": "Search the knowledge base for documents matching a query. Returns ranked results with titles, snippets, and relevance scores. Use this when the user asks a question that requires looking up information from stored documents.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "Natural language search query. Be specific — 'Q4 2025 revenue projections' works better than 'revenue'."
      },
      "limit": {
        "type": "integer",
        "description": "Maximum number of results to return.",
        "default": 10,
        "minimum": 1,
        "maximum" 
      
       
         
         
         
           
             
             
             
          
           
             
                
             
          
        
      
    
     
  

Tool Naming Rules

GOOD tool names (verb_noun, specific):
  search_documents      — clear action + target
  create_github_issue   — includes the service for disambiguation
  get_user_profile      — standard CRUD verb
  analyze_pr_diff       — describes the analysis action
  send_slack_message    — action + channel type

BAD tool names (vague, ambiguous, or too generic):
  search                — search what?
  do_thing              — meaningless
  handler               — not a verb_noun
  processData           — camelCase breaks conventions
  get_stuff             — too vague for LLM selection

Description Engineering

The description is the single most important field for agent tool selection. An LLM reads the description to decide whether to call this tool.

EFFECTIVE description pattern:
"[What it does]. [What it returns]. [When to use it]."

Example:
"Search the knowledge base for documents matching a query. Returns ranked
results with titles, snippets, and relevance scores. Use this when the
user asks a question that requires looking up stored documents."

INEFFECTIVE descriptions:
"Searches documents."           — too short, no usage guidance
"This tool is used for..."      — wastes tokens on filler
"A powerful search engine..."   — marketing copy, not instructions

MCP Server Implementation (TypeScript)

Minimal Server with Tool and Resource

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "project-tools",
  version: "1.0.0",
  capabilities: {
    tools: {},
    resources: {},
  },
});

// Tool: search codebase
server.tool(
  "search_codebase",
  "Search the project codebase for files matching a pattern. Returns file paths and line numbers with matching content. Use when looking for implementations, definitions, or usage of specific code patterns.",
  {
    pattern: z.string().describe("Regex or glob pattern to search for"),
    file_type: z.enum(["ts", "py", "go", "rs", "all"]).default("all")
      .describe("Filter by file extension"),
    max_results: z.number().int().min(1).max(100).default()
      .(),
  },
   ({ pattern, file_type, max_results }) => {
    
     results =  (pattern, file_type, max_results);
     {
      : [{
        : ,
        : .(results, , ),
      }],
    };
  }
);


server.(
  ,
  ,
   (uri) => ({
    : [{
      : uri.,
      : ,
      : .( ()),
    }],
  })
);


 transport =  ();
 server.(transport);

MCP Server with Authentication (SSE Transport)

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from "express";

const app = express();

// Authentication middleware
function authenticateAgent(req, res, next) {
  const token = req.headers.authorization?.replace("Bearer ", "");
  if (!token || !verifyAgentToken(token)) {
    return res.status(401).json({ error: "Invalid agent credentials" });
  }
  req.agentId = extractAgentId(token);
  next();
}

// Rate limiting per agent
const rateLimiter = new Map<string, { count: number; resetAt: number }>();
function rateLimit(agentId: string, maxPerMinute = 60): boolean {
   now = .();
   entry = rateLimiter.(agentId) || { : , : now +  };
   (now > entry.) {
    entry. = ;
    entry. = now + ;
  }
  entry.++;
  rateLimiter.(agentId, entry);
   entry. <= maxPerMinute;
}

app.(, authenticateAgent);

app.(,  {
   (!(req.)) {
     res.().({ :  });
  }
   transport =  (, res);
  server.(transport);
});

app.(,  .());

Google A2A Protocol Implementation

Agent Card (Discovery)

{
  "name": "Research Agent",
  "description": "Performs web research and synthesizes findings into structured reports.",
  "url": "https://research-agent.example.com",
  "provider": {
    "organization": "Acme Corp",
    "url": "https://acme.example.com"
  },
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "stateTransitionHistory": true
  },
  "authentication": {
    "schemes": ["Bearer"],
     
  
   
    
   
    
       
       
       
         
       
        
        
      
    
  

A2A Task Lifecycle

Client                           Agent
  │                                │
  ├─ POST /tasks/send ────────────►│ Create task
  │◄──────── task (submitted) ─────┤
  │                                │
  ├─ GET /tasks/{id} ─────────────►│ Poll status
  │◄──────── task (working) ───────┤
  │                                │
  │  (agent processes...)          │
  │                                │
  ├─ GET /tasks/{id} ─────────────►│ Poll status
  │◄──────── task (completed) ─────┤
  │         + artifacts            │

A2A Client Implementation

import httpx
import json
from dataclasses import dataclass
from typing import Optional
from enum import Enum

class TaskState(Enum):
    SUBMITTED = "submitted"
    WORKING = "working"
    INPUT_REQUIRED = "input-required"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELED = "canceled"

@dataclass
class A2AClient:
    base_url: str
    auth_token: str
    timeout: float = 30.0

    def _headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self.auth_token}",
            "Content-Type": "application/json",
        }

    def discover(self) -> dict:
        """Fetch the agent card for capability discovery."""
        resp = httpx.get(
            f"{self.base_url}/.well-known/agent.json",
            headers=self._headers(),
            timeout=self.timeout,
        )
        resp.raise_for_status()
        return resp.json()

    def send_task() -> :
        
        payload = {
            : ,
            : ,
            : {
                : {
                    : ,
                    : [{: , : message}],
                },
            },
            : task_id  ._generate_id(),
        }
        resp = httpx.post(
            ,
            json=payload,
            headers=._headers(),
            timeout=.timeout,
        )
        resp.raise_for_status()
         resp.json()[]

     () -> :
        
        payload = {
            : ,
            : ,
            : {: task_id},
            : ._generate_id(),
        }
        resp = httpx.post(
            ,
            json=payload,
            headers=._headers(),
            timeout=.timeout,
        )
        resp.raise_for_status()
         resp.json()[]

     () -> :
        
         time
        terminal_states = {TaskState.COMPLETED, TaskState.FAILED, TaskState.CANCELED}
         _  (max_polls):
            task = .get_task(task_id)
             TaskState(task[][])  terminal_states:
                 task
            time.sleep(poll_interval)
         TimeoutError()


     () -> :
         uuid
         (uuid.uuid4())

Protocol Bridge Pattern

When your system uses multiple protocols, implement a bridge that translates between them.

class ProtocolBridge:
    """Translates between MCP tool calls and A2A task delegation."""

    def __init__(self, a2a_agents: dict[str, A2AClient]):
        self.agents = a2a_agents  # skill_id -> A2AClient

    def mcp_tool_to_a2a_task(self, tool_name: str, arguments: dict) -> dict:
        """Convert an MCP tool call into an A2A task send."""
        agent_id, skill = self._resolve_agent(tool_name)
        client = self.agents[agent_id]

        message = self._format_task_message(tool_name, arguments)
        task = client.send_task(message)
        result = client.wait_for_completion(task["id"])

        return self._a2a_result_to_mcp_response(result)

    def _resolve_agent(self, tool_name: str) -> tuple[str, str]:
        """Map MCP tool name to A2A agent + skill."""
        routing = {
            "search_web": ("research-agent", "web-research"),
            "analyze_data": ("analytics-agent", "data-analysis"),
            "generate_code": ("code-agent", "code-generation"),
        }
        if tool_name not in routing:
             ValueError()
         routing[tool_name]

     () -> :
         json.dumps({: tool_name, : arguments})

     () -> :
        
         task[][] == :
            artifacts = task.get(, [])
            text_parts = []
             artifact  artifacts:
                 part  artifact.get(, []):
                     part[] == :
                        text_parts.append(part[])
             {: [{: , : .join(text_parts)}]}
        :
            error_msg = task[].get(, )
             {: [{: , : }], : }

Error Handling Standards

Structured Error Responses

Every protocol should return errors in a consistent format that agents can parse and recover from.

{
  "error": {
    "code": "RATE_LIMITED",
    "message": "Too many requests. Retry after 30 seconds.",
    "retryable": true,
    "retry_after_seconds": 30,
    "details": {
      "limit": 60,
      "window": "1m",
      "current": 62
    }
  }
}

Error Code Taxonomy

Code	Meaning	Agent Action
INVALID_INPUT	Bad parameters	Fix input and retry
NOT_FOUND	Resource missing	Try alternative or report
AUTH_FAILED	Credentials invalid	Refresh token and retry
AUTH_EXPIRED	Token expired	Refresh and retry once
RATE_LIMITED	Too many requests	Wait retry_after then retry
UPSTREAM_ERROR	External service failed	Retry with backoff
INTERNAL_ERROR	Server bug	Report, do not retry
CAPABILITY_UNAVAILABLE	Tool/skill disabled	Use alternative tool

Testing Agent Protocols

Tool Schema Validation

import jsonschema

def validate_mcp_tool(tool_def: dict) -> list[str]:
    """Validate an MCP tool definition for common issues."""
    issues = []

    if not tool_def.get("name"):
        issues.append("Missing tool name")
    elif not tool_def["name"].replace("_", "").isalnum():
        issues.append(f"Tool name '{tool_def['name']}' should use snake_case with alphanumeric chars")

    desc = tool_def.get("description", "")
    if len(desc) < 20:
        issues.append("Description too short — LLMs need clear usage guidance")
    if not any(word in desc.lower() for word in ["use when", "returns", "use this"]):
        issues.append("Description should explain when to use the tool and what it returns")

    schema = tool_def.get("inputSchema", {})
    if schema.get("type") != "object":
        issues.append("inputSchema must be type: object")

    for prop_name, prop_def in schema.get("properties", {}).items():
        if not prop_def.get("description"):
            issues.append()
         prop_def.get() ==    prop_def.get():
            issues.append()

     issues

Integration Testing Pattern

import subprocess
import json

def test_mcp_server_tools():
    """Verify MCP server starts and lists expected tools."""
    proc = subprocess.Popen(
        ["node", "dist/index.js"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    # Send initialize request
    init_msg = json.dumps({
        "jsonrpc": "2.0",
        "method": "initialize",
        "params": {"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "test"}},
        "id": 1,
    }) + "\n"
    proc.stdin.write(init_msg.encode())
    proc.stdin.flush()

    # Send tools/list
    list_msg = json.dumps({
        "jsonrpc": "2.0",
        "method": "tools/list",
        "params": {},
        "id": 2,
    }) + "\n"
    proc.stdin.write(list_msg.encode())
    proc.stdin.flush()

    # Read and validate response
    # (In production, use proper JSON-RPC response parsing)
    proc.terminate()

Common Pitfalls

• Vague tool descriptions that cause the LLM to select the wrong tool or skip it entirely
• Missing required field declarations leading to agents sending incomplete parameters
• No error codes in responses, forcing agents to parse error messages with heuristics
• Exposing internal implementation details in tool schemas instead of user-intent abstractions
• No rate limiting on MCP servers, allowing runaway agent loops to exhaust resources
• Mixing transport concerns with protocol logic instead of keeping them separate
• No capability versioning making it impossible to evolve tools without breaking clients
• Synchronous-only design that blocks on long-running operations instead of using task lifecycle

Best Practices

1. Description-first design — write the tool description before the implementation
2. One intent per tool — a tool that does three things gets selected for the wrong reason
3. Validate inputs on the server — never trust that the LLM sent correct types
4. Return structured errors with codes, not string messages
5. Version your protocol — use capability negotiation at connection time
6. Log every tool call with agent ID, inputs, outputs, and latency for debugging
7. Test tool selection — present your tool list to an LLM and verify it picks the right one
8. Use protocol bridges at boundaries rather than forcing all agents onto one protocol

Troubleshooting

Problem	Cause	Solution
LLM never selects the correct tool	Tool description is vague or missing usage guidance	Rewrite description using the "[What it does]. [What it returns]. [When to use it]." pattern
MCP server connects but no tools appear	Missing tools in server capabilities declaration	Add capabilities: { tools: {} } to the McpServer constructor options
A2A task stuck in working state indefinitely	Agent has no timeout or heartbeat mechanism	Implement max_polls and poll_interval in wait_for_completion; add server-side task TTLs
AUTH_EXPIRED errors after token refresh	Refreshed token not propagated to in-flight requests	Store tokens centrally and read from shared state per-request rather than caching on the client instance
Protocol bridge drops artifacts from A2A responses	Bridge only extracts text parts, ignoring file or data parts	Extend _a2a_result_to_mcp_response to handle all artifact part types including binary and structured data
Rate limiting triggers during normal multi-tool calls	Per-agent rate limit is too low for parallel tool execution	Increase the per-minute ceiling or implement token-bucket rate limiting with burst allowance
Tool schema validation passes but agent sends wrong types	JSON Schema type is correct but lacks format, enum, or pattern constraints	Add tighter constraints (e.g., "format": "date", "pattern": "^[A-Z]{3}$") to catch malformed inputs early

Success Criteria

• Tool selection accuracy >= 95%: LLMs select the intended tool on the first attempt when presented with the full tool list and a matching user query.
• Schema validation coverage = 100%: Every deployed tool passes validate_mcp_tool() with zero issues reported.
• Error response consistency: All protocol endpoints return structured error objects with code, message, and retryable fields — no raw exception strings.
• Discovery latency < 500ms: Agent card retrieval (A2A) and tools/list (MCP) responses complete within 500ms at the 95th percentile.
• Protocol bridge translation fidelity >= 99%: Cross-protocol calls preserve all input parameters and output artifacts without data loss or type coercion errors.
• Authentication failure recovery < 2 retries: Token refresh flows resolve AUTH_EXPIRED errors within a single retry cycle without user intervention.
• Mean time to integrate a new tool < 30 minutes: A developer with access to this skill can define, validate, and deploy a new MCP or A2A tool in under 30 minutes.

Scope & Limitations

This skill covers:

• Designing tool schemas for MCP, A2A, OpenAI Function Calling, and LangChain Tools
• Transport selection, capability discovery, and protocol version negotiation
• Authentication, rate limiting, and structured error handling for agent communication
• Protocol bridging between heterogeneous agent ecosystems

This skill does NOT cover:

• Building complete MCP server applications with business logic — see engineering/mcp-server-builder
• Agent orchestration patterns, planning loops, or multi-step reasoning — see engineering/agent-workflow-designer
• Designing agent personas, memory systems, or behavioral profiles — see engineering/agent-designer
• Infrastructure deployment, CI/CD pipelines, or container orchestration for agent services — see engineering/senior-devops

Integration Points

Skill	Integration	Data Flow
engineering/mcp-server-builder	Protocol schemas defined here feed directly into MCP server scaffolding	Tool definitions and inputSchema objects flow into server code generation
engineering/agent-workflow-designer	Workflow orchestrators consume protocol interfaces to dispatch tasks	Agent-protocol defines the transport contract; workflow-designer defines execution order and branching
engineering/agent-designer	Agent identity and capability profiles reference protocol-level skill declarations	Agent cards and capability metadata from protocol design inform agent persona configuration
engineering/senior-security	Security review of auth flows, token scoping, and rate limiting configurations	OAuth 2.1 flows, API key rotation policies, and audit logging patterns flow into security assessments
engineering/api-design-reviewer	REST and JSON-RPC endpoint design review for A2A and MCP HTTP transports	API schema and endpoint contracts feed into design review checklists
engineering/observability-designer	Monitoring and tracing for inter-agent calls, latency tracking, and error budgets	Tool call logs with agent ID, latency, and error codes flow into observability dashboards