简单Dify调用MCP服务笔记

1. 简单实现一个查询IP信息的MCP服务

mcp服务端，mcp_server.py

import argparse
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, IPvAnyAddress
import ipinfo
import os
import dotenv

dotenv.load_dotenv()

# Create MCP server
mcp = FastMCP("BasicServer")


class IPDetails(BaseModel):
    ip: IPvAnyAddress = None  # type: ignore
    hostname: str | None = None
    city: str | None = None
    region: str | None = None
    country: str | None = None
    loc: str | None = None
    timezone: str | None = None

    class Config:
        extra = "ignore"


@mcp.tool()
async def fetch_ipinfo(ip: str | None = None, **kwargs) -> IPDetails:
    """Get the detailed information of a specified IP address

    Args:
        ip(str or None): The IP address to get information for. Follow the format like 192.168.1.1 .
        **kwargs: Additional keyword arguments to pass to the IPInfo handler.
    Returns:
        IPDetails: The detailed information of the specified IP address.
    """
    handler = ipinfo.getHandler(
        access_token=os.environ.get("IPINFO_API_TOKEN", None), #https://ipinfo.io 获取API_KEY
        headers={"user-agent": "basic-mcp-server", "custom_header": "yes"},
        **kwargs,
    )

    details = handler.getDetails(ip_address=ip)

    return IPDetails(**details.all)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--server_type", type=str, default="sse", choices=["sse", "stdio"])
    args = parser.parse_args()
    mcp.run(args.server_type)

# 终端：启动SSE服务端 python mcp_server.py --server_type sse

mcp客户端，mcp_client.py

from contextlib import asynccontextmanager
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from mcp.client.stdio import StdioServerParameters, stdio_client
from urllib.parse import urlparse
import logging

logger = logging.getLogger(__name__)

class MCPClient:
    def __init__(self, command_or_url: str, args: list[str] = [], env: dict[str, str] = {}):
        self.command_or_url = command_or_url
        self.args = args
        self.env = env

    async def _receive_loop(self, session: ClientSession):
        logger.info("Starting receive loop")
        async for message in session.incoming_messages:
            if isinstance(message, Exception):
                logger.error("Error: %s", message)
                continue
            logger.info("Received message from server: %s", message)

    @asynccontextmanager
    async def _run_session(self):
        if urlparse(self.command_or_url).scheme in ("http", "https"):
            async with sse_client(self.command_or_url) as streams:
                async with ClientSession(*streams) as session:
                    logger.info("Initializing session")
                    await session.initialize()
                    yield session
        else:
            server_parameters = StdioServerParameters(
                command=self.command_or_url, args=self.args, env=self.env
            )
            async with stdio_client(server_parameters) as streams:
                async with ClientSession(*streams) as session:
                    logger.info("Initializing session")
                    await session.initialize()
                    yield session

    async def call_tool(self, tool_name: str, arguments: dict):
        async with self._run_session() as session:
            return await session.call_tool(tool_name, arguments)

    async def list_tools(self):
        async with self._run_session() as session:
            return await session.list_tools()

2. 在Dify中使用MCP服务

2.1在智能体Agent中使用

需要安装插件MCP SSE / StreamableHTTP，在Marketpace里就能搜索到

点击去授权，填写启动的MCP服务地址

{
  "server_name1": {
    "transport": "sse",
    "url": "http://192.168.209.1:8000/sse",
    "headers": {},
    "timeout": 60,
    "sse_read_timeout": 300
  }
}

新建智能体，添加工具