HELP fastapi & MCP(fastAPI .mount etc...) on A2A

[cciwami]

I have reviewed the A2A tutorial and the A2A protocol documentation (https://www.a2aprotocol.org/tutorials/implementing-a2a-in-your-application), but I’m currently facing several challenges:

• There don’t seem to be any examples of integrating A2A with FASTAPI or
  using it alongside MCP (since A2A is not MCP-compliant, they appear to be incompatible).
  How can I add A2A protocol support to an existing external service built with FASTAPI?
  Is such an integration even possible?

• It looks like the agentcard is being created inside each agent folder’s main file.
  Why is this done there instead of defining it in an agent.json file?

• The folder structure across demos, samples, and src appears inconsistent,
  which makes it difficult to understand the recommended or standard layout.

• If I want to expose the system as an API, can the existing demo UI be reused as-is?

Any guidance or examples would be greatly appreciated.

Currently, instead of using MCP, I’m integrating A2A using app.include_router(a2a_router, prefix="/a2a")(Please refer to the following.), which allows the external service and A2A to coexist. Both the external service and A2A are located at the same directory level in the folder structure.

code
`
a2a_src_path = (
Path(file).resolve().parent.parent.parent.parent
/ "A2A"
/ "a2a-python-sdk"
/ "src"
)
sys.path.append(str(a2a_src_path))
a2a_samples_path = (
Path(file).resolve().parent.parent.parent.parent / "A2A" / "samples" / "python"
)
sys.path.append(str(a2a_samples_path))
a2a_demo_path = Path(file).resolve().parent.parent.parent.parent / "A2A" / "demo"
sys.path.append(str(a2a_demo_path))
from a2a.server.fastapi_router import router as a2a_router

a2a_router, prefix="/a2a"
`

A2A apirouter.py
`
import json

from a2a.client.client import A2ACardResolver, A2AClient
from a2a.server.tasks.task_manager import TaskManager
from a2a.types import (
A2ARequest,
AgentCapabilities,
AgentCard,
AgentSkill,
GetTaskPushNotificationConfigRequest,
JSONRPCResponse,
MessageSendParams,
SetTaskPushNotificationConfigRequest,
TaskIdParams,
TaskPushNotificationConfig,
TaskQueryParams,
TaskResubscriptionRequest,
)
from fastapi import APIRouter, Request
from pydantic import ValidationError
from starlette.responses import JSONResponse

router = APIRouter()

class MyTaskManager(TaskManager):
async def dispatch(self, request: A2ARequest) -> JSONRPCResponse:
request_id = request.root.id
method = request.root.method
params = request.root.params

    match method:
        case "tasks/send":
            return await self.on_send_task(params, request_id)
        case "tasks/cancel":
            return await self.on_cancel_task(params, request_id)
        case "tasks/get":
            return await self.on_get_task(params, request_id)
        case "tasks/pushNotification/set":
            return await self.on_set_task_push_notification(params, request_id)
        case "tasks/pushNotification/get":
            return await self.on_get_task_push_notification(params, request_id)
        case "tasks/resubscribe":
            return await self.on_resubscribe_to_task(params, request_id)
        case "tasks/sendSubscribe":
            return await self.on_send_task_subscribe(params, request_id)
        case _:
            return JSONRPCResponse(
                id=request_id,
                error=JSONRPCError(code=-32601, message="Method not found"),
            )

async def on_send_task(
    self, params: MessageSendParams, request_id
) -> JSONRPCResponse:
    return JSONRPCResponse(
        id=request_id,
        result=Message(
            messageId="msg-001",
            parts=params.message.parts,
            role=Role.agent,
            type="message",
        ),
    )

async def on_cancel_task(self, params: TaskIdParams, request_id) -> JSONRPCResponse:
    return JSONRPCResponse(
        id=request_id,
        result=Task(
            id=params.id,
            contextId="ctx-001",
            status={"state": TaskState.canceled},
            type="task",
        ),
    )

async def on_get_task(self, params: TaskQueryParams, request_id) -> JSONRPCResponse:
    return JSONRPCResponse(
        id=request_id,
        result=Task(
            id=params.id,
            contextId="ctx-001",
            status={"state": TaskState.completed},
            type="task",
        ),
    )

async def on_get_task_push_notification(
    self, params: TaskIdParams, request_id
) -> JSONRPCResponse:
    return JSONRPCResponse(
        id=request_id,
        result={
            "taskId": params.id,
            "pushNotificationConfig": {
                "url": "http://localhost:8000/push",
                "token": "abc123",
                "authentication": {
                    "schemes": ["Bearer"],
                    "credentials": "xyz-token",
                },
            },
        },
    )

async def on_set_task_push_notification(
    self, params: TaskPushNotificationConfig, request_id
) -> JSONRPCResponse:
    return JSONRPCResponse(id=request_id, result=params)

async def on_resubscribe_to_task(
    self, params: TaskIdParams, request_id
) -> JSONRPCResponse:
    return JSONRPCResponse(
        id=request_id,
        result=Task(
            id=params.id,
            contextId="ctx-001",
            status={"state": TaskState.working},
            type="task",
        ),
    )

async def on_send_task_subscribe(
    self, params: MessageSendParams, request_id
) -> JSONRPCResponse:
    return JSONRPCResponse(
        id=request_id,
        result=Message(
            messageId="msg-002",
            parts=params.message.parts,
            role=Role.agent,
            type="message",
        ),
    )

task_manager = MyTaskManager()

@router.post("/")
async def handle_rpc(request: Request):
try:
data = await request.json()
rpc_request = A2ARequest.validate_python(data)
result = await task_manager.dispatch(rpc_request)
return JSONResponse(result.model_dump(exclude_none=True))
except ValidationError as e:
return JSONResponse({"error": e.errors()}, status_code=400)
except Exception as e:
return JSONResponse({"error": str(e)}, status_code=500)

@router.get("/.well-known/agent.json")
async def get_agent_card():
return JSONResponse(A2ACardResolver.model_dump(exclude_none=True))

`

[benclarkeio]

Hey @cciwami, you linked an unofficial A2A protocol site, likely generated by AI. That might explain why it doesn't make sense.

Today I released a package called A2A Utils which makes it trivial to use A2A servers. This package also powers the most feature-rich and up-to-date A2A MCP Server.

Let me know if you have any questions!