构建代理式 MLOps：一种基于 A2A 和 MCP 的分层协议策略

#mcp_server.pyfrom mcp.server.fastmcp import FastMCPdef serve(host, port, transport):    """Initializes and runs the MCP Server    Args:        host: The hostname or IP address to bind the server to.        port: The port number to bind the server to.        transport: The transport mechanism for the MCP server (e.g., 'stdio', 'sse').    """    mcp = FastMCP("validation-deployment-mcp-server", host=host, port=port)    @mcp.tool(        name="fetch_model",        description="MCP Tool that fetches the latest trained user churn model.",    )    def fetch_model(model_version_metadata : dict) -> dict:        """MCP Tool that fetches the latest trained user churn model metadata.        Args:            model_version_metadata: Which model data is required.         Returns:            JSON object that returns the Metadata information where the new model            is present and other metadata for validation purposes like test dataset            for validation etc.        """        pass    @mcp.tool(        name="validate_churn_model",        description="MCP that validates the churn model.",    )    def validate_churn_model(validation_config: dict) -> dict:        """MCP Tool that validates the churn model based on validation_config.        Args:            validation_config: config containing validation requirements.        Returns:            JSON object returning the validation status.        """        pass    @mcp.tool(        name="deploy_churn_model",        description="MCP that deploys the churn model.",    )    def deploy_churn_model(deployment_config: dict) -> dict:        """MCP Tool that deploys the churn model based on deployment_config.        Args:            deployment_config: config containing deployment requirements.        Returns:            JSON object returning the deployment status.        """        pass    @mcp.resource("resource://list_agent_cards/list", mime_type="application/json")    def list_agent_cards() -> dict:        """Retrieves all loaded agent cards as a json / dictionary for the MCP resource endpoint.        This function serves as the handler for the MCP resource identified by        the URI 'resource://agent_cards/list'.        Returns:            A JSON object containing a list of all available agents.        """    @mcp.resource(        "resource://retrieve_agent_skills/{agent_name}", mime_type="application/json"    )    def retrieve_agent_skills(agent_name: str) -> dict:        """Retrieves an agent card as JSON data.        Returns:            A JSON object of Agent Card.        """        pass    mcp.run(transport=transport)def main(host, port, transport) -> None:    serve(host, port, transport)

from contextlib import asynccontextmanagerfrom typing import Any, AsyncGenerator, Dict, Listfrom mcp import ClientSessionfrom mcp.types import ReadResourceResult, ListResourcesResult, ListToolsResultclass MCPClient:    """A high-level client for interacting with the MLOps MCP server."""    def __init__(self, host: str, port: int, transport: str):        """        Initializes the client with the server's connection details.                Args:            host: The hostname or IP of the MCP server.            port: The port of the MCP server.            transport: The transport mechanism (e.g., 'http', 'sse').        """        self._host = host        self._port = port        self._transport = transport    @asynccontextmanager    async def _get_session(self) -> AsyncGenerator[ClientSession, None]:        """        Provides a managed session to connect with the MCP server.        The actual implementation of this would depend on the chosen transport.        """        # In a real implementation, you would initialize the session here        # based on self._host, self._port, etc.        session: ClientSession = None  # Placeholder for the actual session object        try:            # For example:connected through http            yield session        finally:            # For example: await session.close()            pass    async def list_agents(self) -> ReadResourceResult:        """        Retrieves the list of all available agent cards from the MCP server.        """        async with self._get_session() as session:            return await session.read_resource("resource://list_agent_cards/list")    async def get_agent_skills(self, agent_name: str) -> ReadResourceResult:        """        Retrieves the skills for a specific agent from the MCP server.        """        async with self._get_session() as session:            uri = f"resource://retrieve_agent_skills/{agent_name}"            return await session.read_resource(uri)    async def list_resources(self) -> ListResourcesResult:        """Lists all available resources on the MCP server."""        async with self._get_session() as session:            return await session.list_resources()    async def list_tools(self) -> ListToolsResult:        """Lists all available tools on the MCP server."""        async with self._get_session() as session:            return await session.list_tools()

import jsonfrom collections.abc import AsyncIterablefrom a2a.client import A2AClientfrom uuid import uuid4import httpxfrom a2a.types import (    AgentCard,    MessageSendParams,    SendStreamingMessageRequest,    SendStreamingMessageSuccessResponse,    TaskArtifactUpdateEvent,)from mcp_client import MCPClientfrom a2a.server.agent_execution import AgentExecutor, RequestContextfrom a2a.server.events import EventQueueclass Task:    """Represents a single task that needs to be executed in the task list."""    task_query: str    def __init__(self, *args, **kwargs):        pass    async def find_agent_for_task(self, mcp_client, query) -> AgentCard | None:        """Fetch an agent card suitable for the node's task from MCP."""        result = await mcp_client.list_agents(query)        chosen_agent = select_agent(query)        agent_card_json = json.loads(chosen_agent.content[0].text)        return AgentCard(**agent_card_json)    async def execute_task(        self,    ) -> AsyncIterable[dict[str, any]]:        """Execute the node task via A2A streaming messages using the assigned agent."""        agent_card = await self.find_agent_for_task(query=self.task_query)        async with httpx.AsyncClient() as httpx_client:            client = A2AClient(httpx_client, agent_card)  # A2A Client queries the Agent            payload: dict[str, any] = {                "message": {                    "parts": [{"kind": "text", "text": self.task_query}],                    # Can have other elements too based on Agent Card inputs.                },            }            request = SendStreamingMessageRequest(                id=str(uuid4()), params=MessageSendParams(**payload)            )            response_stream = client.send_message_streaming(request)            async for chunk in response_stream:                # Save the artifact as a result of the node                if isinstance(chunk.root, SendStreamingMessageSuccessResponse) and                              isinstance(chunk.root.result, TaskArtifactUpdateEvent):                    artifact = chunk.root.result.artifact                    self.results = artifact                yield chunkclass TaskList:    """Represents a Topological graph of tasks that need to be executed"""    task_list: list[Task]  # Task list that needs to be executed.    def __init__(self, *args, **kwargs) -> None:        """        Breaks the query into a task list and the order in which it should be         executed.The AI agent should break this down and put it in the task_list         array.        """        pass    async def execute_task_list(self) -> AsyncIterable[dict[str, any]]:        """        Executes the tasks for the agent.        """        # .....        for task in self.task_list:            # ....            task.execute_task()class GenericAgentExecutor(AgentExecutor):    """AgentExecutor used by the agents."""    def __init__(self, agent):        self.agent = agent    async def execute(        self,        context: RequestContext,        event_queue: EventQueue,    ) -> None:        pass

{    "name": "Orchestrator Agent",    "description": "Helps in invoking the MLOps workflow. Which will do validtion and deployment",    "url": "http://localhost:8003/",    "version": "1.0.0",    "skills": [        {            "id": "orchestrate_the_flow",            "name": "orchestrate_the_flow",            "description": "Helps in orchestrating MLOps Workflow",            "tags": [                "Validate the model and then deploy it."            ],            "examples": [                "Retrieve the latest churn prediction model and run it through the validation module. If the model’s absolute bias is less than or equal to 0.04, approve it for deployment. Deploy the new model to the alternate region: if the current production model is running in us-west-1, deploy this version to us-west-2; otherwise, deploy it to us-west-1."            ]        }    ]}

from typing import AsyncIterable, Anyfrom agent_helpers import TaskListfrom mcp_client import MCPClientclass OrchestratorAgent:    """    Orchestrates a multi-step workflow by breaking a high-level goal    into a sequence of tasks for specialist agents.    """    def __init__(self, mcp_client: MCPClient, prompt_personality: str):        """        Initializes the Orchestrator Agent.        Args:            mcp_client: A client for interacting with the MCP server.            prompt_personality: Instructions guiding the agent's planning process.        """        self._mcp_client = mcp_client        self._prompt_personality = prompt_personality    async def _create_plan_from_query(self, query: str) -> TaskList:        """        Translates a natural language query into a structured TaskList for delegation.        """        # This method simulates the agent's high-level reasoning process.        # The agent's LLM, guided by its personality prompt, would parse the        # user's query to identify distinct, sequential steps.        # For our example query, it would identify two main sub-goals:        # 1. A validation step with a specific condition.        # 2. A deployment step that depends on the outcome of the first.        # The agent then creates a TaskList where each Task encapsulates the        # natural language instruction for that sub-goal. This is different        # from a specialist agent, whose plan would involve specific tool calls.        #        # Task 1 Query: "Retrieve the latest churn prediction model... approve it for deployment."        # Task 2 Query: "Deploy the new model to the alternate region..."        #        # The output of this method would be a TaskList object containing        # these two Task objects, ready for execution.        pass    async def stream(self, query: str) -> AsyncIterable[dict[str, Any]]:        """        Processes a query by creating a plan and then executing it.        """        # 1. CREATE THE PLAN        # The agent first calls its internal planning method to translate        # the natural language query into a structured TaskList.        plan = await self._create_plan_from_query(query)        # 2. EXECUTE THE PLAN        # The agent then executes the plan. The plan.execute() method will        # iterate through the Tasks. For each Task, it will find the        # appropriate specialist agent (Validation, then Deployment) and        # stream the sub-query to it. The results are then yielded back.        pass

{    "name": "Validation Agent",    "description": "Helps in validating the MLOps model.",    "url": "http://localhost:8004/",    "version": "1.0.0",    "skills": [        {            "id": "validate_the_model",            "name": "validate_the_model",            "description": "Helps in validating MLOps models",            "tags": [                "Validate the model based on user requirements."            ],            "examples": [                "Retrieve the latest churn prediction model and run it through the validation module. If the model’s absolute bias is less than or equal to 0.04, approve it for deployment."            ]        }    ]}

from typing import AsyncIterable, Anyfrom mcp_client import MCPClientclass ValidationAgent:    """    A specialist agent that validates a machine learning model by discovering    and using tools from the MCP server.    """    def __init__(self, mcp_client: MCPClient, prompt_personality: str):        """        Initializes the Validation Agent.        Args:            mcp_client: A client for interacting with the MCP server.            prompt_personality: Instructions guiding the agent's tool-use logic.        """        self._mcp_client = mcp_client        self._prompt_personality = prompt_personality    async def _create_tool_use_plan(self, query: str):        """        Translates a natural language query into a structured plan of tool calls.        """        # This method simulates the agent's reasoning process.        # 1. DISCOVER: The agent first needs to understand what it can do.        # It would call self._mcp_client.list_tools() to get a real-time        # list of all available capabilities on the MCP server. This allows        # it to dynamically learn that tools like 'fetch_model' and        # 'validate_churn_model' are available. This should be part of the         # prompt_personality        # 2. PLAN: Based on the available tools and the specific user query,        # the agent formulates a plan. For the query: "...absolute bias is        # less than or equal to 0.04...", its LLM would determine that it        # needs to:        #   a. Fetch the model's metadata using the 'fetch_model' tool.        #   b. Construct a 'validation_config' containing the bias check,        #      extracting the '0.04' threshold from the query.        #   c. Call the 'validate_churn_model' tool with that config.        #        # The output of this method would be a structured object, like a list        # of pre-configured tool calls, ready for execution.        pass    async def stream(self, query: str) -> AsyncIterable[dict[str, Any]]:        """        Processes a validation query by creating a plan and then executing it.        """        # 1. CREATE THE PLAN        # The agent first calls its internal planning method to translate        # the natural language query into a structured sequence of tool calls.        plan = await self._create_tool_use_plan(query)        # 2. EXECUTE THE PLAN        # The agent would then iterate through the steps in the generated plan.        # It would call the necessary MCP client methods (fetch_model,        # validate_churn_model) in the correct order with the correct        # parameters derived during the planning phase. The results of each        # step would be yielded back to the Orchestrator.        pass

{    "name": "Deployment Agent",    "description": "Helps in deploying the validated MLOps model.",    "url": "http://localhost:8005/",    "version": "1.0.0",    "skills": [        {            "id": "deploy_the_model",            "name": "deploy_the_model",            "description": "Helps in deploying MLOps models",            "tags": [                "Deploy the model based on user requirements."            ],            "examples": [                "Deploy the new model to the alternate region: if the current production model is running in us-west-1, deploy this version to us-west-2; otherwise, deploy it to us-west-1."            ]        }    ]}

from typing import AsyncIterable, Anyfrom mcp_client import MCPClientclass DeploymentAgent:    """    A specialist agent that deploys a validated machine learning model by    discovering and using tools from the MCP server.    """    def __init__(self, mcp_client: MCPClient, prompt_personality: str):        """        Initializes the Deployment Agent.        Args:            mcp_client: A client for interacting with the MCP server.            prompt_personality: Instructions guiding the agent's tool-use logic.        """        self._mcp_client = mcp_client        self._prompt_personality = prompt_personality    async def _create_tool_use_plan(self, query: str):        """        Translates a natural language query into a structured plan of tool calls.        """        # This method simulates the agent's reasoning process.        # 1. DISCOVER: The agent determines its available capabilities.        # It would call self._mcp_client.list_tools() to learn that tools        # like 'fetch_model' and 'deploy_churn_model' are available. Again done by prompt        # personality.        # 2. PLAN: The agent formulates a plan based on the query: "Deploy        # the new model to the alternate region...". Its LLM reasoning would be:        #   a. To find the "alternate" region, I must first find the "current" one.        #   b. The 'fetch_model' tool can get me the metadata of the current        #      production model.        #   c. From that metadata, I can extract the current deployment region.        #   d. I can then write logic to determine the alternate region.        #   e. The final plan is a sequence of two tool calls: first fetch_model        #      to get the state, then deploy_churn_model to execute the change.        # The output of this method would be a structured object, like a list        # of pre-configured tool calls, ready for execution.        pass    async def stream(self, query: str) -> AsyncIterable[dict[str, Any]]:        """        Processes a deployment query by creating a plan and then executing it.        """        # 1. CREATE THE PLAN        # The agent first calls its internal planning method to translate        # the natural language query into a structured sequence of tool calls.        plan = await self._create_tool_use_plan(query)        # 2. EXECUTE THE PLAN        # The agent would then iterate through the steps in the generated plan.        # It would call the necessary MCP client methods (fetch_model,        # deploy_churn_model) in the correct order with the correct        # parameters derived during the planning phase. The results of each        # step would be yielded back to the caller.        pass

import jsonimport httpxfrom pathlib import Pathfrom basic_helper.promp_personalities import promptsfrom orchestrator_agent import OrchestratorAgentfrom validation_agent import ValidationAgentfrom deployment_agent import DeploymentAgentfrom a2a.types import AgentCardimport uvicornfrom a2a.server.apps import A2AStarletteApplicationfrom a2a.server.request_handlers import DefaultRequestHandlerfrom a2a.server.tasks import (    BasePushNotificationSender,    InMemoryPushNotificationConfigStore,    InMemoryTaskStore,)mcp_client = MCPClient(host="localhost", port=8000, transport="http") # Example Clientdef get_agent(agent_card: AgentCard):    """Get the agent, given an agent card."""    try:        if agent_card.name == "Orchestrator Agent":            # This is the Orchestrator Agent            return OrchestratorAgent(mcp_client, prompts.orchestrator_agent)        if agent_card.name == "Validation Agent":            # This is the Validation Agent            return ValidationAgent(mcp_client, prompts.validation_agent)        if agent_card.name == "Deployment Agent":            # This is the Deployment Agent            return DeploymentAgent(mcp_client, prompts.deployment_agent)    except Exception as e:        raise edef main(host, port, agent_card_path):    """Starts an Agent server."""    with Path.open(agent_card) as file:        data = json.load(file)    agent_card = AgentCard(**data)    client = httpx.AsyncClient()    push_notification_config_store = InMemoryPushNotificationConfigStore()    push_notification_sender = BasePushNotificationSender(        client, config_store=push_notification_config_store    )    request_handler = DefaultRequestHandler(        agent_executor=GenericAgentExecutor(agent=get_agent(agent_card)),        task_store=InMemoryTaskStore(),        push_config_store=push_notification_config_store,        push_sender=push_notification_sender,    )    server = A2AStarletteApplication(        agent_card=agent_card, http_handler=request_handler    )    uvicorn.run(server.build(), host=host, port=port)if __name__ == "__main__":    main()