Agent-to-Agent (A2A) & Multi-MCP Mesh on Cloud Run with Gemma 4, Google ADK, Spanner Graph & AlloyDB

from typing import List
from pydantic import BaseModel, Field


class AgentSkillCard(BaseModel):
    """Skill advertised inside an A2A /.well-known/agent.json AgentCard."""

    id: str = Field(description="Unique A2A skill identifier.")
    name: str = Field(description="Human-readable skill name.")
    description: str = Field(description="Capability summary used by the root ADK coordinator.")
    tags: List[str] = Field(description="Routing tags for horizontal A2A delegation.")


class A2AAgentCard(BaseModel):
    """Standard Agent-to-Agent (A2A) discovery card served at /.well-known/agent.json."""

    name: str = Field(description="Remote ADK specialist agent name.")
    url: str = Field(description="Private Cloud Run HTTPS endpoint for JSON-RPC tasks/send.")
    version: str = Field(default="1.0.0", description="A2A agent contract version.")
    authentication: str = Field(default="google-cloud-run-oidc", description="Required IAM auth scheme.")
    skills: List[AgentSkillCard] = Field(description="Advertised A2A capabilities.")


class IncidentMeshManifest(BaseModel):
    """Final validated remediation manifest produced by the A2A + Multi-MCP Mesh."""

    manifest_id: str = Field(description="Deterministic incident remediation manifest ID.")
    incident_id: str = Field(description="Incoming production incident identifier.")
    root_service_id: str = Field(description="Failing service identified in Spanner Graph.")
    impacted_services: List[str] = Field(description="Downstream services discovered via 3-hop GQL query.")
    matched_runbook_id: str = Field(description="AlloyDB pgvector ScaNN matched runbook ID.")
    a2a_delegated_agent: str = Field(description="Remote A2A specialist agent invoked via tasks/send.")
    github_rollback_pr_url: str = Field(description="GitHub Pull Request URL created via GitHub MCP.")
    rollout_strategy: str = Field(description="Canary traffic split enforced by the LoopAgent safety gate.")
    loop_iterations_completed: int = Field(description="Number of LoopAgent verification passes.")
    final_safety_score: float = Field(description="Final LoopAgent safety score (1.0).")
    model_name: str = Field(description="Gemma 4 model identifier served on Cloud Run GPU.")

from typing import Any, Dict
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("CloudSpannerGraphMCP")

SPANNER_GQL_QUERY = """
GRAPH ServiceDependencyGraph
MATCH (root:Microservice {service_id: @service_id})-[:CALLS]->{1,3}(downstream:Microservice)
      -[:SERVES]->(tenant:EnterpriseTenant)
RETURN root.last_commit_sha AS failing_commit_sha,
       ARRAY_AGG(DISTINCT downstream.service_id) AS impacted_services,
       COUNT(DISTINCT tenant.tenant_id) AS impacted_tenants,
       SUM(DISTINCT tenant.hourly_sla_penalty_usd) AS sla_exposure_usd
"""


@mcp.tool()
def query_service_blast_radius_gql(
    service_id: str,
    max_hops: int = 3,
) -> Dict[str, Any]:
    """Executes a 3-hop ISO GQL graph query against Cloud Spanner Graph to compute incident blast radius."""
    return {
        "root_service_id": service_id,
        "gql_executed": SPANNER_GQL_QUERY.strip(),
        "failing_commit_sha": "8f4d2a1c9e7b",
        "hop_count": max_hops,
        "impacted_downstream_services": [
            "checkout-orchestrator-v2",
            "payment-ledger-writer",
            "eu-vat-settlement-worker",
        ],
        "impacted_enterprise_tenants": 42,
        "sla_breach_risk_usd": 184000.0,
        "query_latency_ms": 11.4,
    }


if __name__ == "__main__":
    mcp.run(transport="sse")

from typing import Any, Dict
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("AlloyDBVectorRunbookMCP")

ALLOYDB_SCANN_SQL = """
SELECT runbook_id,
       1 - (embedding <=> @query_embedding) AS similarity_score,
       remediation_strategy,
       required_safety_checks
FROM sre_verified_runbooks
WHERE service_tier = 'TIER_1_FINANCIAL'
ORDER BY embedding <=> @query_embedding
LIMIT 1;
"""


@mcp.tool()
def search_alloydb_runbooks_scann(
    error_signature: str,
    service_tier: str = "TIER_1_FINANCIAL",
) -> Dict[str, Any]:
    """Queries AlloyDB pgvector with the ScaNN index to retrieve verified SRE remediation runbooks."""
    return {
        "error_signature": error_signature,
        "service_tier": service_tier,
        "sql_executed": ALLOYDB_SCANN_SQL.strip(),
        "runbook_id": "RB-SPANNER-GRPC-POOL-204",
        "similarity_score": 0.964,
        "remediation_strategy": (
            "Revert gRPC channel pool keepalive timeout from 500ms back to 15000ms "
            "and shift traffic via a staged 10% -> 50% -> 100% Cloud Run revision canary."
        ),
        "required_safety_checks": [
            "Prohibit 100% immediate traffic cutover when impacted_enterprise_tenants > 10",
            "Require Cloud Run canary traffic split (10% initial stage) and automated rollback PR",
        ],
        "scann_query_latency_ms": 6.8,
    }


if __name__ == "__main__":
    mcp.run(transport="sse")

from typing import Any, Dict
from fastapi import FastAPI
from .schemas import A2AAgentCard, AgentSkillCard

app = FastAPI(title="A2A Remote GitHub Remediation Agent")

AGENT_CARD = A2AAgentCard(
    name="GitHubRollbackRemediationAgent",
    url="https://github-remediation-a2a-agent.run.internal",
    version="1.0.0",
    authentication="google-cloud-run-oidc",
    skills=[
        AgentSkillCard(
            id="prepare-rollback-pr-and-canary",
            name="Prepare GitHub Rollback PR and Cloud Run Canary Plan",
            description=(
                "Inspects failing commit SHAs via GitHub MCP, opens a verified revert PR, "
                "and proposes a Cloud Run revision rollout spec."
            ),
            tags=["github-mcp", "a2a", "cloud-run", "sre-remediation"],
        )
    ],
)


@app.get("/.well-known/agent.json")
def get_a2a_agent_card() -> Dict[str, Any]:
    """Serves the standard A2A discovery manifest for horizontal agent discovery."""
    return AGENT_CARD.model_dump()


@app.post("/tasks/send")
def handle_a2a_task_send(payload: Dict[str, Any]) -> Dict[str, Any]:
    """Handles JSON-RPC 2.0 A2A tasks/send requests from the root ADK coordinator."""
    params = payload.get("params", {})
    failing_sha = params.get("failing_commit_sha", "8f4d2a1c9e7b")
    target_service = params.get("service_id", "grpc-edge-gateway")
    return {
        "jsonrpc": "2.0",
        "id": payload.get("id", "a2a-task-9041"),
        "result": {
            "task_status": "COMPLETED",
            "delegated_agent": AGENT_CARD.name,
            "github_mcp_tool_invoked": "create_pull_request",
            "github_rollback_pr_url": "https://github.com/RobinaMirbahar/a2a-mcp-mesh-cloudrun/commits/main",
            "reverted_commit_sha": failing_sha,
            "target_service": target_service,
            "proposed_rollout_strategy": "IMMEDIATE_100_PERCENT_CUTOVER",
        },
    }

import os
from google.adk.agents import LlmAgent, LoopAgent, ParallelAgent, SequentialAgent
from google.adk.models.lite_llm import LiteLlm

from .mcp_servers.alloydb_vector_mcp import search_alloydb_runbooks_scann
from .mcp_servers.spanner_graph_mcp import query_service_blast_radius_gql
from .schemas import IncidentMeshManifest
from .coordinator_agent import (
    audit_mesh_remediation_safety,
    delegate_github_remediation_via_a2a,
    exit_safety_audit_loop,
)


def build_a2a_mcp_mesh_coordinator() -> SequentialAgent:
    """Constructs the hybrid Multi-MCP + A2A ParallelAgent -> LoopAgent -> Publisher ADK coordinator."""
    model = LiteLlm(model=os.getenv("MESH_MODEL_NAME", "openai/google/gemma-4-27b-it"))

    spanner_graph_specialist = LlmAgent(
        name="SpannerGraphBlastRadiusAgent",
        model=model,
        instruction="Call `query_service_blast_radius_gql` and store the graph blast radius in `{spanner_blast_radius}`.",
        tools=[query_service_blast_radius_gql],
        output_key="spanner_blast_radius",
    )

    alloydb_runbook_specialist = LlmAgent(
        name="AlloyDBRunbookVectorAgent",
        model=model,
        instruction="Call `search_alloydb_runbooks_scann` and store the matched SRE runbook in `{alloydb_runbook}`.",
        tools=[search_alloydb_runbooks_scann],
        output_key="alloydb_runbook",
    )

    a2a_github_specialist = LlmAgent(
        name="A2AGitHubRemediationClientAgent",
        model=model,
        instruction="Discover the remote A2A agent via `/.well-known/agent.json` and call `delegate_github_remediation_via_a2a`.",
        tools=[delegate_github_remediation_via_a2a],
        output_key="a2a_remediation",
    )

    parallel_mesh_fanout = ParallelAgent(
        name="ConcurrentMultiMCPAndA2AFanOut",
        sub_agents=[spanner_graph_specialist, alloydb_runbook_specialist, a2a_github_specialist],
    )

    safety_critic = LlmAgent(
        name="RemediationSafetyCriticAgent",
        model=model,
        instruction="Call `audit_mesh_remediation_safety`. When `safety_score` reaches 1.0, call `exit_safety_audit_loop`.",
        tools=[audit_mesh_remediation_safety, exit_safety_audit_loop],
        output_key="safety_audit_verdict",
    )

    safety_loop = LoopAgent(
        name="MeshRemediationSafetyLoop",
        sub_agents=[safety_critic],
        max_iterations=3,
    )

    manifest_publisher = LlmAgent(
        name="IncidentMeshManifestPublisher",
        model=model,
        instruction="Synthesize state into a validated `IncidentMeshManifest`.",
        output_key="published_mesh_manifest",
        output_schema=IncidentMeshManifest,
    )

    return SequentialAgent(
        name="CloudRunA2AMCPMeshCoordinator",
        sub_agents=[parallel_mesh_fanout, safety_loop, manifest_publisher],
    )