Python调用大型模型LLM实现mcp服务的实例代码展示

开启mcp服务，这里以网页抓取为例

npx -y supergateway –stdio "uvx mcp-server-fetch"

python代码

import asyncio

import os

from openai import AsyncOpenAI

from mcp.client.sse import sse_client

from mcp import ClientSession

from typing import Optional

from contextlib import AsyncExitStack

import json

class MCPAgent:

    def __init__(self):

        self.exit_stack = AsyncExitStack()

        self.session: Optional[ClientSession] = None

        self.llm = AsyncOpenAI(

            api_key="sk-xxx",

            base_url="https://api.siliconflow.cn/v1/"

        )

    async def connect_server(self, sse_endpoint: str):

        transport = await self.exit_stack.enter_async_context(

            sse_client(sse_endpoint)

        )

        self.session = await self.exit_stack.enter_async_context(

            ClientSession(*transport)

        )

        await self.session.initialize()

        print("可用工具:", [t.name for t in (await self.session.list_tools()).tools])

    async def execute_workflow(self, query: str) -> str:

        # 步骤1：获取工具描述

        tools = [{

            "type": "function",

            "function": {

                "name": t.name,

                "description": t.description,

                "parameters": t.inputSchema

            }

        } for t in (await self.session.list_tools()).tools]

        # 步骤2：LLM决策工具调用

        chat_completion = await self.llm.chat.completions.create(

            model="deepseek-ai/DeepSeek-V3", #deepseek-ai/DeepSeek-R1-Distill-Qwen-7B deepseek-ai/DeepSeek-R1-Distill-Qwen-7B deepseek-ai/DeepSeek-R1

            messages=[{"role": "user", "content": query}],

            tools=tools,

            tool_choice="auto"

        )

        # 步骤3：执行工具调用

        if tool_calls := chat_completion.choices[0].message.tool_calls:

            results = []

            for tool_call in tool_calls:

                args = json.loads(tool_call.function.arguments)

                result = await self.session.call_tool(

                    tool_call.function.name,

                    args

                )

                print(result.content[0].text)

                print("———")

                results.append(result.content[0].text)

            # 步骤4：结果整合

            final_response = await self.llm.chat.completions.create(

                model="deepseek-r1-distill-qwen", #deepseek-ai/DeepSeek-R1-Distill-Qwen-7B

                messages=[{

                    "role": "user",

                    "content": f"原始问题：{query}\n工具结果：{json.dumps(results)}"

                }]

            )

            return final_response.choices[0].message.content

        return chat_completion.choices[0].message.content

        # return json.dumps(results)

async def main():

    agent = MCPAgent()

    try:

        # 连接本地SSE服务

        await agent.connect_server("http://localhost:8000/sse")

        # 示例查询

        response = await agent.execute_workflow(

            "看下这个页面: https://blog.csdn.net/u010479989/article/details/147422815,并总结三个关键点"

        )

        print(f"\n最终响应：{response}")

    finally:

        await agent.exit_stack.aclose()

if __name__ == "__main__":

    asyncio.run(main())

执行结果

作者：代码先觉