### Background Server Start Usage

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Example of starting the server in the background.

```python
server = CustomA2AServer(agent_card, task_manager)
server.start()  # Returns immediately
```

--------------------------------

### Asynchronous Server Start Usage

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Example of starting the server asynchronously.

```python
server = CustomA2AServer(agent_card, task_manager)
await server.start_async()
```

--------------------------------

### A2AClient Initialization Examples

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Examples showing how to instantiate the client using a URL, an AgentCard, or a custom timeout.

```python
# Initialize with URL
client = A2AClient(url="http://localhost:41242")

# Initialize with AgentCard
card = AgentCard(name="MyAgent", url="http://localhost:41242", ...)
client = A2AClient(agent_card=card)

# With custom timeout
client = A2AClient(url="http://localhost:41242", timeout=120.0)
```

--------------------------------

### Complete Claude Desktop Configuration Example

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

A full example of the structure required for the claude_desktop_config.json file.

```json
{
  "mcpServers": {
    "a2a": {
      "command": "uvx",
      "args": [
        "a2a-mcp-server"
      ]
    }
  }
}
```

--------------------------------

### Local Installation

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Clone the repository and set up a local development environment.

```bash
git clone https://github.com/GongRzhe/A2A-MCP-Server.git
cd A2A-MCP-Server
```

```bash
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
```

```bash
pip install -r requirements.txt
```

--------------------------------

### Install from PyPI

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Install the package directly from the Python Package Index.

```bash
pip install a2a-mcp-server
```

--------------------------------

### Start A2A Server Instance

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Basic usage pattern to instantiate and start the A2A server.

```python
server = setup_a2a_server()
server.start()
```

--------------------------------

### Background Server Start Method

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Defines the method to start the server as a background task.

```python
def start(self)
```

--------------------------------

### View server startup output

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Example console output showing configuration details upon startup.

```text
MCP Bridge Configuration:
- Transport: stdio
- Host: 0.0.0.0
- Port: 8000
A2A Server Configuration:
- Host: 0.0.0.0
- Port: 41241
Data Storage:
- Registered Agents: ./registered_agents.json
- Task Agent Mapping: ./task_agent_mapping.json
```

--------------------------------

### Fetch Agent Card Usage Example

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Demonstrates retrieving and printing agent metadata.

```python
card = await fetch_agent_card("http://localhost:41242")
print(f"Agent: {card.name}")
print(f"Capabilities: {card.capabilities}")
```

--------------------------------

### Asynchronous Server Start Method

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Defines the asynchronous start method for the server.

```python
async def start_async(self)
```

--------------------------------

### Install via Smithery

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Automatically install the A2A Bridge Server for Claude Desktop using the Smithery CLI.

```bash
npx -y @smithery/cli install @GongRzhe/A2A-MCP-Server --client claude
```

--------------------------------

### Configure Production Deployment

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Environment setup and directory preparation for persistent production storage.

```bash
#!/bin/bash
export MCP_TRANSPORT="streamable-http"
export MCP_HOST="0.0.0.0"
export MCP_PORT="8000"
export MCP_PATH="/mcp"
export A2A_HOST="0.0.0.0"
export A2A_PORT="41241"
export A2A_MCP_DATA_DIR="/var/lib/a2a-mcp/data"

# Ensure data directory exists
mkdir -p /var/lib/a2a-mcp/data
chmod 700 /var/lib/a2a-mcp/data

# Run server (typically in systemd service or Docker container)
python a2a_mcp_server.py
```

--------------------------------

### Retrieve Task Example

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Demonstrates initializing the task manager and fetching a specific task with history constraints.

```python
task_manager = InMemoryTaskManager()

request = GetTaskRequest(
    params=TaskQueryParams(
        id="task-123",
        historyLength=5
    )
)

response = await task_manager.on_get_task(request)
if response.result:
    print(f"Task state: {response.result.status.state}")
```

--------------------------------

### Example Usage of get_task_callback

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Demonstrates initializing the client and retrieving callback configuration for a task ID.

```python
client = A2AClient(url="http://localhost:41242")

response = await client.get_task_callback({
    "id": "task-123"
})

if response.result:
    print(f"Callback URL: {response.result.pushNotificationConfig.url}")
```

--------------------------------

### Send Task Example

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Example usage of the send_task method to communicate with an agent.

```python
client = A2AClient(url="http://localhost:41242")

message = Message(
    role="user",
    parts=[TextPart(text="What's the weather?")]
)

response = await client.send_task({
    "id": "task-123",
    "message": message,
    "sessionId": "session-456"
})

print(response.result.status.state)  # "completed"
print(response.result.artifacts)      # List of artifacts
```

--------------------------------

### Setup Streamable-HTTP for Web Clients

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Environment variable configuration for web-based clients like Cursor or Windsurf.

```bash
#!/bin/bash
export MCP_TRANSPORT="streamable-http"
export MCP_HOST="127.0.0.1"
export MCP_PORT="8000"
export MCP_PATH="/mcp"
export A2A_MCP_DATA_DIR="./data"

python a2a_mcp_server.py
```

--------------------------------

### Claude MCP Client Interaction Example

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Example sequence of tool calls using Claude as an MCP client to register an agent, send a message, and retrieve results.

```text
User: Register an agent at http://localhost:41242

Claude uses: register_agent(url="http://localhost:41242")
Claude: Successfully registered agent: ReimbursementAgent

User: Ask the agent what it can do

Claude uses: send_message(agent_url="http://localhost:41242", message="What can you do?")
Claude: I've sent your message. Here's the task_id: b30f3297-e7ab-4dd9-8ff1-877bd7cfb6b1

User: Get the answer to my question

Claude uses: get_task_result(task_id="b30f3297-e7ab-4dd9-8ff1-877bd7cfb6b1")
Claude: The agent replied: "I can help you process reimbursement requests. Just tell me what you need to be reimbursed for, including the date, amount, and purpose."
```

--------------------------------

### Configure Server via Environment Variables

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Example of setting environment variables for transport and network configuration before launching the server.

```bash
export MCP_TRANSPORT="streamable-http"
export MCP_HOST="127.0.0.1"
export MCP_PORT="8000"
export A2A_HOST="0.0.0.0"
export A2A_PORT="41241"
python a2a_mcp_server.py
```

--------------------------------

### Run Configuration Creator

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Execute the automated setup script to generate or update the Claude Desktop configuration.

```bash
python config_creator.py
```

--------------------------------

### Usage of on_send_task_subscribe

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Example of initiating a task subscription and iterating over the resulting event stream.

```python
request = SendTaskStreamingRequest(
    params=TaskSendParams(
        id="task-123",
        message=Message(...),
        metadata={"agent_url": "http://localhost:41242"}
    )
)

response = await task_manager.on_send_task_subscribe(request)
async for event in response:
    if event.result:
        print(f"Update: {event.result}")
```

--------------------------------

### Configure Claude Desktop via Local Installation

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Use this configuration block when running from a local repository clone.

```json
"a2a": {
  "command": "C:\\path\\to\\python.exe",
  "args": [
    "C:\\path\\to\\A2A-MCP-Server\\a2a_mcp_server.py"
  ],
  "env": {
    "MCP_TRANSPORT": "stdio",
    "PYTHONPATH": "C:\\path\\to\\A2A-MCP-Server"
  }
}
```

--------------------------------

### Configure push notifications

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Example usage of the set_push_notification_info method to associate a callback configuration with a task.

```python
config = PushNotificationConfig(
    url="https://myapp.com/callback",
    token="secret-token"
)

await task_manager.set_push_notification_info("task-123", config)
```

--------------------------------

### Configure Claude Desktop for A2A MCP

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

JSON configuration for Claude Desktop using either PyPI or local installation paths.

```json
{
  "mcpServers": {
    "a2a": {
      "command": "uvx",
      "args": ["a2a-mcp-server"]
    }
  }
}
```

```json
{
  "mcpServers": {
    "a2a": {
      "command": "python",
      "args": ["/path/to/a2a_mcp_server.py"],
      "env": {
        "MCP_TRANSPORT": "stdio",
        "PYTHONPATH": "/path/to/A2A-MCP-Server"
      }
    }
  }
}
```

--------------------------------

### Usage of _forward_task_stream

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Example of invoking the internal forwarding task via asyncio.

```python
# Usually called internally by on_send_task_subscribe
asyncio.create_task(task_manager._forward_task_stream(client, request, task_id))
```

--------------------------------

### Store Values with Cache Set

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Examples of storing data with and without TTL, and overwriting existing keys.

```python
cache = InMemoryCache()

# Store indefinitely
cache.set("persistent", "data")

# Store with 1-hour expiration
cache.set("session", "token", ttl=3600)

# Overwrite existing
cache.set("key", "new_value", ttl=1800)
```

--------------------------------

### Invoke send_message_stream

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/01-mcp-tools.md

Example usage of the send_message_stream function to request a story from an agent.

```python
result = await send_message_stream(
    agent_url="http://localhost:41242",
    message="Tell me a story",
    ctx=context
)
# Returns: {
#     "status": "success",
#     "task_id": "xyz789",
#     "messages": ["Once upon a time...", "...in a land far away..."],
#     "artifacts": []
# }
```

--------------------------------

### Registered Agents Storage Format

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Example structure for the registered_agents.json file containing agent metadata.

```json
{
  "http://localhost:41242": {
    "url": "http://localhost:41242",
    "name": "ReimbursementAgent",
    "description": "Processes expense reimbursement requests"
  },
  "http://localhost:41243": {
    "url": "http://localhost:41243",
    "name": "CurrencyAgent",
    "description": "Provides currency conversion rates"
  }
}
```

--------------------------------

### Fetch and display agent card

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Example usage of A2ACardResolver to retrieve and print agent details.

```python
resolver = A2ACardResolver("http://localhost:41242")
card = resolver.get_agent_card()
print(f"Agent: {card.name}")
print(f"Capabilities: {card.capabilities}")
```

--------------------------------

### Start A2A MCP Server with HTTP Transport

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Command to launch the server using the streamable-http transport for client connectivity.

```bash
MCP_TRANSPORT=streamable-http MCP_HOST=127.0.0.1 MCP_PORT=8000 uvx a2a-mcp-server
```

--------------------------------

### Configure task callback using A2AClient

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Defines the method signature and provides an example of setting a push notification callback for a specific task.

```python
async def set_task_callback(
    self, payload: dict[str, Any]
) -> SetTaskPushNotificationResponse
```

```python
{
    "id": str,                              # Task ID
    "pushNotificationConfig": {             # Callback configuration
        "url": str,                         # Callback URL
        "token": str,                       # Optional callback token
        "authentication": {...}             # Optional auth config
    }
}
```

```python
client = A2AClient(url="http://localhost:41242")

response = await client.set_task_callback({
    "id": "task-123",
    "pushNotificationConfig": {
        "url": "https://myapp.com/callback",
        "token": "secret-token"
    }
})
```

--------------------------------

### Task Agent Mapping Format

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Example structure for the task_agent_mapping.json file mapping task IDs to agent URLs.

```json
{
  "b30f3297-e7ab-4dd9-8ff1-877bd7cfb6b1": "http://localhost:41242",
  "550e8400-e29b-41d4-a716-446655440000": "http://localhost:41243",
  "abc123def456ghi789": "http://localhost:41242"
}
```

--------------------------------

### Execute cancel_task tool

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/01-mcp-tools.md

Example usage of the cancel_task function to terminate a specific task.

```python
result = await cancel_task(
    task_id="b30f3297-e7ab-4dd9-8ff1-877bd7cfb6b1",
    ctx=context
)
# Returns: {
#     "status": "success",
#     "task_id": "b30f3297-e7ab-4dd9-8ff1-877bd7cfb6b1",
#     "message": "Task cancelled successfully"
# }
```

--------------------------------

### Handle A2AClientHTTPError

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/08-error-handling.md

Example usage for catching and inspecting HTTP errors during client communication.

```python
try:
    client = A2AClient(url="http://localhost:41242")
    response = await client.send_task(payload)
except A2AClientHTTPError as e:
    print(f"HTTP {e.status_code}: {e.message}")
    if e.status_code == 503:
        print("Agent is temporarily unavailable")
```

--------------------------------

### get(key: str, default: Any = None) -> Any

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Retrieves a value from the cache by key. If the key is missing or expired, the provided default value is returned.

```APIDOC
## get(key: str, default: Any = None) -> Any

### Description
Retrieves a value from the cache. This method is thread-safe, checks TTL, and removes expired entries.

### Parameters
- **key** (str) - Required - Cache key
- **default** (Any) - Optional - Value to return if not found or expired

### Returns
- **Any** - The cached value or the default value if the key is missing or expired.
```

--------------------------------

### Invoke get_task_result

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/01-mcp-tools.md

Example usage of the get_task_result function with a specific task ID and history length.

```python
result = await get_task_result(
    task_id="b30f3297-e7ab-4dd9-8ff1-877bd7cfb6b1",
    history_length=5,
    ctx=context
)
# Returns: {
#     "status": "success",
#     "task_id": "b30f3297-e7ab-4dd9-8ff1-877bd7cfb6b1",
#     "state": "completed",
#     "message": "Your request has been processed",
#     "history": [...],
#     "artifacts": [...]
# }
```

--------------------------------

### Consume SSE events from task manager

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Example usage of dequeue_events_for_sse to iterate over task responses and handle stream completion.

```python
queue = await task_manager.setup_sse_consumer("task-123")

async for response in task_manager.dequeue_events_for_sse(
    request_id=req_id,
    task_id="task-123",
    sse_event_queue=queue
):
    print(f"Received: {response.result}")
    if response.result and hasattr(response.result, 'final') and response.result.final:
        print("Stream complete")
```

--------------------------------

### Handle A2AClientJSONError

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/08-error-handling.md

Example usage for catching JSON parsing errors.

```python
try:
    response = await client.send_task(payload)
except A2AClientJSONError as e:
    print(f"Invalid JSON from agent: {e.message}")
```

--------------------------------

### Register and Query Agent Workflow

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/09-api-workflows.md

Demonstrates the three-step process of registering an agent, sending a message, and polling for the final task result.

```python
# Step 1: Register the agent
result = await register_agent(
    url="http://localhost:41242",
    ctx=context
)

if result["status"] == "success":
    agent_name = result["agent"]["name"]
    print(f"Registered: {agent_name}")
else:
    print(f"Registration failed: {result['message']}")
    exit(1)

# Step 2: Send a message to the agent
response = await send_message(
    agent_url="http://localhost:41242",
    message="What can you do?",
    ctx=context
)

if response["status"] == "success":
    task_id = response["task_id"]
    print(f"Message sent, task ID: {task_id}")
    print(f"State: {response['state']}")
    if "message" in response:
        print(f"Response: {response['message']}")
else:
    print(f"Send failed: {response['message']}")
    exit(1)

# Step 3: Get full result (if not immediate)
final_result = await get_task_result(
    task_id=task_id,
    ctx=context
)

if final_result["status"] == "success":
    print(f"Final state: {final_result['state']}")
    print(f"Message: {final_result.get('message')}")
else:
    print(f"Retrieval failed: {final_result['message']}")
```

--------------------------------

### setup_a2a_server

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Factory function to create and configure the A2A server instance.

```APIDOC
## setup_a2a_server

### Description
Factory function to create and configure the A2A server. It initializes the AgentCard, sets up capabilities, creates a task manager, and binds to the configured host and port.

### Signature
`def setup_a2a_server() -> CustomA2AServer`

### Returns
- **CustomA2AServer** - Configured server instance.

### Example
```python
server = setup_a2a_server()
server.start()
```
```

--------------------------------

### Define main function signature

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

The synchronous entry point that triggers the application startup.

```python
def main()
```

--------------------------------

### setup_sse_consumer

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Sets up a queue for streaming task updates via SSE for a specific task ID.

```APIDOC
## setup_sse_consumer

### Description
Sets up a queue for streaming task updates via SSE. Creates a new queue and registers it for the task.

### Method
async def setup_sse_consumer(self, task_id: str, is_resubscribe: bool = False)

### Parameters
- **task_id** (str) - Required - Task ID to subscribe to
- **is_resubscribe** (bool) - Optional - Whether this is a resubscription

### Returns
- **asyncio.Queue** - Queue for receiving events.
```

--------------------------------

### Initialize and Use InMemoryCache

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Demonstrates singleton behavior and basic set/get operations.

```python
# Get singleton instance
cache = InMemoryCache()
cache.set("key1", "value1")
value = cache.get("key1")

# Subsequent calls return same instance
cache2 = InMemoryCache()
assert cache is cache2
```

--------------------------------

### Execute task cancellation

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Example usage of the on_cancel_task method to cancel a specific task by ID.

```python
request = CancelTaskRequest(
    params=TaskIdParams(id="task-123")
)

response = await task_manager.on_cancel_task(request)
if response.error:
    print(f"Error: {response.error.message}")
```

--------------------------------

### Define setup_sse_consumer signature

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Defines the asynchronous method signature for initializing an SSE consumer queue.

```python
async def setup_sse_consumer(
    self, task_id: str, is_resubscribe: bool = False
)
```

--------------------------------

### Configure Claude Desktop via PyPI

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Add this entry to the mcpServers section of the Claude Desktop configuration file.

```json
"a2a": {
  "command": "uvx",
  "args": [
    "a2a-mcp-server"
  ]
}
```

--------------------------------

### A2ACardResolver.__init__

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Initializes the A2ACardResolver with a base URL and an optional path to the agent card.

```APIDOC
## A2ACardResolver.__init__

### Description
Initializes a new instance of the A2ACardResolver class.

### Parameters
- **base_url** (str) - Required - Agent base URL
- **agent_card_path** (str) - Optional - Path to agent card (default: /.well-known/agent.json)
```

--------------------------------

### Define main_async function signature

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

The primary asynchronous entry point for initializing the server components.

```python
async def main_async()
```

--------------------------------

### Cancel a task using A2AClient

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Defines the method signature and provides an example of cancelling a task by ID.

```python
async def cancel_task(self, payload: dict[str, Any]) -> CancelTaskResponse
```

```python
{
    "id": str,                              # Task ID to cancel
    "metadata": dict,                       # Optional metadata
}
```

```python
client = A2AClient(url="http://localhost:41242")

response = await client.cancel_task({
    "id": "task-123"
})

if response.error:
    print(f"Cancel failed: {response.error.message}")
else:
    print("Task cancelled successfully")
```

--------------------------------

### Load JWKS configuration

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Method to initialize the JWKS client and fetch public keys from a specified URL.

```python
async def load_jwks(self, jwks_url: str)
```

```python
auth = PushNotificationReceiverAuth()
await auth.load_jwks("https://agent.example.com/.well-known/jwks.json")
```

--------------------------------

### Register an A2A agent

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/01-mcp-tools.md

Defines the signature for registering an agent and provides a usage example for the register_agent function.

```python
async def register_agent(url: str, ctx: Context) -> Dict[str, Any]
```

```python
result = await register_agent(url="http://localhost:41242", ctx=context)
# Returns: {
#     "status": "success",
#     "agent": {
#         "url": "http://localhost:41242",
#         "name": "ReimbursementAgent",
#         "description": "Handles reimbursement requests"
#     }
# }
```

--------------------------------

### Configure Local Development Environment

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Set variables for stdio transport and debug logging to facilitate local development.

```bash
export MCP_TRANSPORT="stdio"
export A2A_MCP_DATA_DIR="./local-data"
export FASTMCP_LOG_LEVEL="DEBUG"
```

--------------------------------

### Deploy with Docker

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Dockerfile and execution command for containerized deployment.

```dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

ENV MCP_TRANSPORT=streamable-http
ENV MCP_HOST=0.0.0.0
ENV MCP_PORT=8000
ENV A2A_HOST=0.0.0.0
ENV A2A_PORT=41241
ENV A2A_MCP_DATA_DIR=/data

VOLUME ["/data"]
EXPOSE 8000

CMD ["python", "a2a_mcp_server.py"]
```

```bash
docker run -p 8000:8000 -v ~/a2a-data:/data a2a-mcp-server
```

--------------------------------

### Verify Push Notification URL

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Validates a callback URL by sending a GET request with a random token.

```python
@staticmethod
async def verify_push_notification_url(url: str) -> bool
```

```python
auth = PushNotificationSenderAuth()
is_valid = await auth.verify_push_notification_url("https://callback.example.com")
if is_valid:
    print("URL is reachable and valid")
```

--------------------------------

### A2AClient.__init__

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Initializes a new A2AClient instance to communicate with an A2A agent.

```APIDOC
## A2AClient.__init__(agent_card=None, url=None, timeout=60.0)

### Description
Initializes the A2A client. Either agent_card or url must be provided; if both are provided, agent_card takes precedence.

### Parameters
- **agent_card** (AgentCard) - Optional - Agent card object containing the agent URL.
- **url** (str) - Optional - Direct agent URL string.
- **timeout** (TimeoutTypes) - Optional - Request timeout in seconds (default: 60.0).

### Errors
- Raises ValueError if neither agent_card nor url is provided.
```

--------------------------------

### CustomA2AServer Constructor

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Initializes the A2A server with an agent card and task manager.

```python
class CustomA2AServer:
    def __init__(
        self,
        agent_card: AgentCard,
        task_manager: A2ABridgeTaskManager,
        host: str = "0.0.0.0",
        port: int = 41241,
    )
```

--------------------------------

### Define AgentSkill Model

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/02-types.md

Describes a specific capability offered by an agent, including input/output modes and usage examples.

```python
class AgentSkill(BaseModel):
    id: str
    name: str
    description: str | None = None
    tags: list[str] | None = None
    examples: list[str] | None = None
    inputModes: list[str] | None = None
    outputModes: list[str] | None = None
```

--------------------------------

### Configure Low Memory Environment

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Set the data directory to a specific path to manage memory and storage usage.

```bash
# Use minimal task history
# Configure smaller data directory
export A2A_MCP_DATA_DIR="/tmp/a2a-cache"
```

--------------------------------

### Find Agent by Capability

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/09-api-workflows.md

Shows how to list available agents and filter them based on their description to find a suitable agent for a specific task.

```python
# Get list of agents
agents = await list_agents()

if not agents:
    print("No agents registered")
    exit(1)

print(f"Found {len(agents)} agents:")
for agent in agents:
    print(f"  - {agent['name']}: {agent['description']}")

# Find agents that might handle financial queries
financial_agents = [
    a for a in agents 
    if "financial" in a.get("description", "").lower() 
    or "currency" in a.get("description", "").lower()
]

if financial_agents:
    agent_url = financial_agents[0]["url"]
    print(f"Using financial agent: {agent_url}")
    
    response = await send_message(
        agent_url=agent_url,
        message="Show me the S&P 500 index",
        ctx=context
    )
else:
    print("No financial agents found")
```

--------------------------------

### List registered A2A agents

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/01-mcp-tools.md

Defines the signature for listing agents and provides a usage example for the list_agents function.

```python
async def list_agents() -> List[Dict[str, Any]]
```

```python
agents = await list_agents()
# Returns: [
#     {
#         "url": "http://localhost:41242",
#         "name": "ReimbursementAgent",
#         "description": "Handles reimbursement requests"
#     }
# ]
```

--------------------------------

### Initialize A2A Server

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Factory function signature for creating a configured CustomA2AServer instance.

```python
def setup_a2a_server() -> CustomA2AServer
```

--------------------------------

### Consume SSE task events

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Demonstrates how to initialize a consumer queue and process incoming events until a final event is received.

```python
queue = await task_manager.setup_sse_consumer("task-123")

# In another coroutine, events are enqueued:
# await task_manager.enqueue_events_for_sse("task-123", event)

# Consume events:
while True:
    event = await queue.get()
    print(f"Event type: {type(event)}")
    if hasattr(event, 'final') and event.final:
        break
```

--------------------------------

### Implement Factory Pattern for A2A Clients

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/10-architecture.md

Creates or retrieves A2A client instances based on the provided agent URL.

```python
def get_or_create_client(self, agent_url: str) -> A2AClient:
    if agent_url not in self.agent_clients:
        self.agent_clients[agent_url] = A2AClient(url=agent_url)
    return self.agent_clients[agent_url]
```

--------------------------------

### verify_push_notification_url(url: str) -> bool

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Verifies that a push notification callback URL is valid and reachable by sending a GET request with a validation token.

```APIDOC
## verify_push_notification_url

### Description
Verifies that a push notification URL is valid and reachable. It sends a GET request with a random validation token and expects the response text to match the token.

### Parameters
- **url** (str) - Required - URL to verify

### Returns
- **bool** - True if URL responds correctly to validation request, False otherwise.
```

--------------------------------

### Basic A2A Workflow

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

The sequential steps for interacting with an A2A agent via an MCP client.

```text
1. Client registers an A2A agent
   ↓
2. Client sends a message to the agent (gets task_id)
   ↓
3. Client retrieves the task result using task_id
```

--------------------------------

### Troubleshooting Commands

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Commands used to diagnose port conflicts, file permissions, and agent registration issues.

```bash
lsof -i :8000
```

```bash
ls -la ./registered_agents.json
```

```bash
python /full/path/to/a2a_mcp_server.py
```

```bash
curl http://agent-url/
```

```bash
curl http://agent-url/.well-known/agent.json
```

--------------------------------

### Configure A2A Server Binding

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Sets the host and port for the A2A metadata server.

```bash
export A2A_HOST="127.0.0.1"
export A2A_PORT="41241"
```

--------------------------------

### are_modalities_compatible(server_output_modes, client_output_modes)

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Checks if server and client output modes are compatible. Returns True if compatible, False otherwise.

```APIDOC
## are_modalities_compatible(server_output_modes, client_output_modes)

### Description
Checks if server and client output modes are compatible. Compatibility is defined as either list being empty/None, or having at least one common mode.

### Parameters
- **server_output_modes** (list[str]) - Required - Supported output modes
- **client_output_modes** (list[str]) - Required - Requested output modes

### Returns
- **bool** - True if compatible, False otherwise.
```

--------------------------------

### Project Directory Structure

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Visual representation of the project file hierarchy.

```text
a2a-mcp-server/
├── a2a_mcp_server.py      # Main server implementation
├── common/                # A2A protocol code (from google/A2A)
│   ├── client/            # A2A client implementation
│   ├── server/            # A2A server implementation
│   ├── types.py           # Common type definitions
│   └── utils/             # Utility functions
├── config_creator.py      # Script to help create Claude Desktop configuration
├── .gitignore             # Git ignore file
├── pyproject.toml         # Project metadata and dependencies
├── README.md              # This file
└── requirements.txt       # Project dependencies
```

--------------------------------

### Usage of are_modalities_compatible

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Demonstrates compatibility checks between server and client output modes.

```python
from common.server.utils import are_modalities_compatible

# Compatible
are_modalities_compatible(["text/plain", "application/json"], ["text/plain"])  # True
are_modalities_compatible(["text/plain"], [])  # True

# Not compatible
are_modalities_compatible(["text/plain"], ["image/png"])  # False
```

--------------------------------

### Configure High Throughput Environment

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Set environment variables to enable streamable HTTP transport for better scalability and parallel request handling.

```bash
export MCP_TRANSPORT="streamable-http"
export MCP_HOST="0.0.0.0"
export MCP_PORT="8000"
```

--------------------------------

### Process Agent Artifacts in Python

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/09-api-workflows.md

Demonstrates the workflow of sending a message, retrieving task results, and iterating through artifacts to save them as text or JSON files.

```python
response = await send_message(
    agent_url="http://localhost:41242",
    message="Generate a CSV report of expenses",
    ctx=context
)

if response["status"] != "success":
    exit(1)

# Get task result with artifacts
result = await get_task_result(
    task_id=response["task_id"],
    ctx=context
)

# Process artifacts
if "artifacts" in result:
    for artifact in result["artifacts"]:
        print(f"Artifact: {artifact['name']}")
        
        for part in artifact.get("contents", []):
            if part["type"] == "text":
                # Save text artifact
                filename = f"{artifact['name']}.txt"
                with open(filename, "w") as f:
                    f.write(part["text"])
                print(f"  Saved to {filename}")
            
            elif part["type"] == "data":
                # Process structured data
                import json
                filename = f"{artifact['name']}.json"
                with open(filename, "w") as f:
                    json.dump(part["data"], f)
                print(f"  Saved to {filename}")
```

--------------------------------

### Configure MCP Server Binding

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Sets the host address, port, and transport protocol for HTTP-based MCP communication.

```bash
export MCP_HOST="127.0.0.1"
export MCP_PORT="8080"
export MCP_TRANSPORT="streamable-http"
```

--------------------------------

### Run A2A MCP Server via Command Line

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Execute the server using stdio or HTTP transport modes.

```bash
# Using default settings (stdio transport)
uvx a2a-mcp-server

# Using HTTP transport on specific host and port
MCP_TRANSPORT=streamable-http MCP_HOST=127.0.0.1 MCP_PORT=8080 uvx a2a-mcp-server
```

--------------------------------

### Execute A2A MCP server

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Command to launch the server from the command line.

```bash
python a2a_mcp_server.py
```

--------------------------------

### Initialize A2ACardResolver

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Constructor for the A2ACardResolver class, requiring a base URL for the agent.

```python
class A2ACardResolver:
    def __init__(self, base_url, agent_card_path='/.well-known/agent.json')
```

--------------------------------

### Initialize A2AClient

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Constructor definition for the A2AClient class.

```python
class A2AClient:
    def __init__(
        self,
        agent_card: AgentCard = None,
        url: str = None,
        timeout: TimeoutTypes = 60.0,
    )
```

--------------------------------

### Configure Environment Variables

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Set environment variables to define transport types, host, port, and logging levels.

```bash
# Transport type: stdio, streamable-http, or sse
export MCP_TRANSPORT="streamable-http"

# Host for the MCP server
export MCP_HOST="0.0.0.0"

# Port for the MCP server (when using HTTP transports)
export MCP_PORT="8000"

# Path for the MCP server endpoint (when using HTTP transports)
export MCP_PATH="/mcp"

# Path for SSE endpoint (when using SSE transport)
export MCP_SSE_PATH="/sse"

# Enable debug logging
export MCP_DEBUG="true"
```

--------------------------------

### Usage of new_not_implemented_error

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Demonstrates creating an UnsupportedOperationError response.

```python
from common.server.utils import new_not_implemented_error

response = new_not_implemented_error("req-123")
# Returns: JSONRPCResponse(
#     id="req-123",
#     error=UnsupportedOperationError()
# )
```

--------------------------------

### Usage of new_incompatible_types_error

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Demonstrates creating a ContentTypeNotSupportedError response.

```python
from common.server.utils import new_incompatible_types_error

response = new_incompatible_types_error("req-123")
# Returns: JSONRPCResponse(
#     id="req-123",
#     error=ContentTypeNotSupportedError()
# )
```

--------------------------------

### Retrieve Task Callback Configuration

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Defines the method signature for fetching push notification settings for a specific task.

```python
async def get_task_callback(
    self, payload: dict[str, Any]
) -> GetTaskPushNotificationResponse
```

--------------------------------

### Configure Data Storage Directory

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/07-configuration.md

Sets the directory path for persistent storage of agent and task mapping files.

```bash
export A2A_MCP_DATA_DIR="/var/lib/a2a-mcp-server"
python a2a_mcp_server.py
```

--------------------------------

### Implement Observer Pattern for Task Streaming

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/10-architecture.md

Uses a queue-based pub-sub mechanism for handling SSE subscriptions and event broadcasting.

```python
# Setup: Create queue for task
queue = await setup_sse_consumer(task_id="task-123")

# Publish: Broadcast event to all subscribers
await enqueue_events_for_sse(task_id="task-123", event)

# Subscribe: Consume events from queue
async for response in dequeue_events_for_sse(queue):
    # Process update
```

--------------------------------

### Retrieve or create A2AClient

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Fetches an existing A2AClient from the cache or creates a new one if it does not exist.

```python
def get_or_create_client(self, agent_url: str) -> A2AClient
```

```python
task_manager = A2ABridgeTaskManager()
client = task_manager.get_or_create_client("http://localhost:41242")
# Subsequent calls return the same instance
client2 = task_manager.get_or_create_client("http://localhost:41242")
assert client is client2
```

--------------------------------

### load_jwks(jwks_url: str)

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Loads JWKS from a specified public keys endpoint and initializes the client for key fetching and caching.

```APIDOC
## load_jwks(jwks_url: str)

### Description
Loads JWKS from a public keys endpoint. Creates a PyJWKClient that fetches and caches keys, with automatic refreshing.

### Parameters
- **jwks_url** (str) - Required - URL to JWKS endpoint

### Example
```python
auth = PushNotificationReceiverAuth()
await auth.load_jwks("https://agent.example.com/.well-known/jwks.json")
```
```

--------------------------------

### get_or_create_client

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Retrieves an existing A2AClient for a specific agent URL or creates a new one if it does not exist.

```APIDOC
## get_or_create_client

### Description
Get or create an A2AClient for a specific agent. Checks cache first for existing client and creates a new one if not cached.

### Parameters
- **agent_url** (str) - Required - URL of the A2A agent

### Returns
- **A2AClient** - The client instance for the agent.

### Example
```python
task_manager = A2ABridgeTaskManager()
client = task_manager.get_or_create_client("http://localhost:41242")
```
```

--------------------------------

### CustomA2AServer

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

A class for initializing and managing an A2A server instance.

```APIDOC
## CustomA2AServer

### Description
Initializes a minimal A2A server implementation for the bridge.

### Constructor Parameters
- **agent_card** (AgentCard) - Required - Card describing this server.
- **task_manager** (A2ABridgeTaskManager) - Required - Task manager instance.
- **host** (str) - Optional - Bind address (default: "0.0.0.0").
- **port** (int) - Optional - Bind port (default: 41241).

### Methods
- **start_async()**: Starts the server asynchronously and runs indefinitely.
- **start()**: Starts the server as a background task and returns immediately.
```

--------------------------------

### Specify Transport Type

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/README.md

Configure the transport protocol using environment variables or command-line prefixes.

```bash
# Using environment variable
export MCP_TRANSPORT="streamable-http"
uvx a2a-mcp-server

# Or directly in the command
MCP_TRANSPORT=streamable-http uvx a2a-mcp-server
```

--------------------------------

### Define FileContent model

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/02-types.md

Contains file metadata and content; either bytes or uri must be provided.

```python
class FileContent(BaseModel):
    name: str | None = None
    mimeType: str | None = None
    bytes: str | None = None
    uri: str | None = None
```

--------------------------------

### Define get_agent_card method

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Method signature for synchronously fetching an agent card.

```python
def get_agent_card(self) -> AgentCard
```

--------------------------------

### Check modality compatibility

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Defines the function signature for checking if server and client output modes are compatible.

```python
def are_modalities_compatible(
    server_output_modes: list[str], client_output_modes: list[str]
) -> bool
```

--------------------------------

### Retrieve task push notification

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Demonstrates how to request and access the push notification configuration for a specific task.

```python
request = GetTaskPushNotificationRequest(
    params=TaskIdParams(id="task-123")
)

response = await task_manager.on_get_task_push_notification(request)
if response.result:
    print(f"Callback URL: {response.result.pushNotificationConfig.url}")
```

--------------------------------

### tasks/get

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/02-types.md

Retrieves the current state of a task.

```APIDOC
## JSON-RPC Method: tasks/get

### Description
Retrieves the current state information for a specific task.

### Parameters
- **params** (TaskQueryParams) - Required - The parameters required to query the task state.
```

--------------------------------

### Check for push notification configuration

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Verifies the existence of a callback configuration for a given task ID.

```python
async def has_push_notification_info(self, task_id: str) -> bool
```

```python
has_config = await task_manager.has_push_notification_info("task-123")
if has_config:
    config = await task_manager.get_push_notification_info("task-123")
```

--------------------------------

### Retrieve push notification configuration

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Fetches the configuration for a specific task. Raises a ValueError if the task ID is not found.

```python
async def get_push_notification_info(
    self, task_id: str
) -> PushNotificationConfig
```

```python
config = await task_manager.get_push_notification_info("task-123")
print(f"Callback URL: {config.url}")
```

--------------------------------

### Implement Singleton Pattern in Python

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/10-architecture.md

Ensures a single global instance of the cache using thread-safe locking.

```python
class InMemoryCache:
    _instance: Optional['InMemoryCache'] = None
    _lock: threading.Lock = threading.Lock()
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance
```

--------------------------------

### Define on_get_task Method

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Method signature for retrieving a task by ID.

```python
async def on_get_task(self, request: GetTaskRequest) -> GetTaskResponse
```

--------------------------------

### Define Task Model

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/02-types.md

Represents a complete task including history, artifacts, and metadata.

```python
class Task(BaseModel):
    id: str
    sessionId: str | None = None
    status: TaskStatus
    artifacts: list[Artifact] | None = None
    history: list[Message] | None = None
    metadata: dict[str, Any] | None = None
```

--------------------------------

### tasks/send

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/02-types.md

Sends a task to an agent.

```APIDOC
## JSON-RPC Method: tasks/send

### Description
Sends a task to an agent for processing.

### Parameters
- **params** (TaskSendParams) - Required - The parameters required to send the task.
```

--------------------------------

### Send Multiple Queries Concurrently

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/09-api-workflows.md

Demonstrates sending messages to multiple agents and polling for their results asynchronously.

```python
import asyncio

agents = [
    "http://localhost:41242",  # Finance agent
    "http://localhost:41243",  # Weather agent
]

queries = [
    "What's the USD/EUR exchange rate?",
    "What's the weather forecast for tomorrow?",
]

# Send all messages
tasks = []
for agent_url, query in zip(agents, queries):
    response = await send_message(
        agent_url=agent_url,
        message=query,
        ctx=context
    )
    if response["status"] == "success":
        tasks.append({
            "agent": agent_url,
            "task_id": response["task_id"],
            "query": query,
            "response": response
        })

# Poll for results
results = {}
for task in tasks:
    result = await get_task_result(
        task_id=task["task_id"],
        ctx=context
    )
    results[task["agent"]] = {
        "query": task["query"],
        "answer": result.get("message"),
        "state": result.get("state")
    }

# Use results
for agent_url, data in results.items():
    print(f"Agent: {agent_url}")
    print(f"  Query: {data['query']}")
    print(f"  Answer: {data['answer']}")
    print()
```

--------------------------------

### Implement Structured Error Responses in MCP Tools

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/08-error-handling.md

Use try-except blocks to catch specific exceptions and return a consistent dictionary structure with status and message fields.

```python
@mcp.tool()
async def my_tool(param: str) -> Dict[str, Any]:
    try:
        # Do work
        result = await do_something(param)
        return {
            "status": "success",
            "data": result
        }
    except ValueError as e:
        return {
            "status": "error",
            "message": f"Validation failed: {str(e)}"
        }
    except Exception as e:
        return {
            "status": "error",
            "message": f"Unexpected error: {str(e)}"
        }
```

--------------------------------

### Define send_task Method

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/03-a2a-client.md

Method signature for sending tasks to an agent.

```python
async def send_task(self, payload: dict[str, Any]) -> SendTaskResponse
```

--------------------------------

### enqueue_events_for_sse

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Broadcasts an event to all subscribers of a specific task.

```APIDOC
## enqueue_events_for_sse

### Description
Broadcasts an event to all subscribers of a task. If no subscribers exist, the operation does nothing.

### Method
async def enqueue_events_for_sse(self, task_id, task_update_event)

### Parameters
- **task_id** (str) - Required - Task ID
- **task_update_event** (TaskStatusUpdateEvent | TaskArtifactUpdateEvent) - Required - Event to broadcast
```

--------------------------------

### Define enqueue_events_for_sse signature

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Defines the asynchronous method signature for broadcasting task updates to subscribers.

```python
async def enqueue_events_for_sse(self, task_id, task_update_event)
```

--------------------------------

### Fetch Agent Card Function Definition

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/05-mcp-bridge-server.md

Defines the signature for fetching agent metadata from a URL.

```python
async def fetch_agent_card(url: str) -> AgentCard
```

--------------------------------

### Define set_push_notification_info method signature

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/04-task-manager.md

Defines the asynchronous interface for storing push notification configuration for a task.

```python
async def set_push_notification_info(
    self, task_id: str, notification_config: PushNotificationConfig
)
```

--------------------------------

### Initialize PushNotificationReceiverAuth

Source: https://github.com/gongrzhe/a2a-mcp-server/blob/main/_autodocs/06-utilities.md

Class definition for handling receiver-side push notification authentication.

```python
class PushNotificationReceiverAuth(PushNotificationAuth):
    def __init__(self)
```