FastMCP（Streamable HTTP）とLiteLLMでツール呼び出し可能な対話アプリをお試しで作る

import requests
from fastmcp import FastMCP

mcp = FastMCP("Sample MCP Server", "これはテスト用のMCPサーバーです")  # type: ignore


@mcp.tool()
def get_address_by_postal_code(postal_code: str) -> dict:
    """
    郵便番号から住所情報を取得する。

    Parameters
    ----------
    postal_code : str
        住所を取得したい郵便番号（7桁）

    Returns
    -------
    dict
        住所情報（都道府県、市区町村、町域など）を含む辞書。該当なしの場合は空の辞書。
    """
    API_URL = "https://zipcloud.ibsnet.co.jp/api/search"
    params = {"zipcode": postal_code}
    try:
        response = requests.get(API_URL, params=params, timeout=5)
        response.raise_for_status()
        result = response.json()
        if result["status"] == 200 and result["results"]:
            return result["results"][0]
        else:
            return {}
    except Exception:
        # エラー時は空の辞書を返す
        return {}


if __name__ == "__main__":
    mcp.run(transport="streamable-http")

python sample_mcp_server.py

import asyncio

from fastmcp import Client


async def main():

    # MCPクライアントの準備
    client = Client("http://127.0.0.1:8000/mcp")

    async with client:
        # MCPサーバで提供されているツール一覧の取得
        tools = await client.list_tools()
        print(f"Available tools: {tools}")


if __name__ == "__main__":
    asyncio.run(main())

from typing import Any
from mcp import Tool


def fastmcp_tools_to_litellm_tools(tools: list[Tool]) -> list[dict[str, Any]]:
    """
    fastmcpのTool情報をlitellm function call用tools dict形式に変換する。

    Parameters
    ----------
    tools : list[Tool]
        fastmcpで取得したToolオブジェクトのリスト

    Returns
    -------
    list[dict[str, Any]]
        litellm function call用tools(dict)リスト
    """
    return [
        {
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                # 実装により inputSchema / input_schema の差異があるため両対応
                "parameters": getattr(tool, "inputSchema", getattr(tool, "input_schema", None)),
            },
        }
        for tool in tools
    ]

import litellm
from litellm.types.utils import ModelResponse
from mcp import Tool
from typing import Any


def call_llm_with_tools(
    messages: list[dict[str, Any]], tools: list[Tool], model: str = "azure/gpt-4o-mini"
) -> ModelResponse:
    """
    メッセージ履歴とツール情報を渡してLLMのfunction call出力を得る。

    Parameters
    ----------
    messages : list[dict[str, Any]]
        チャット履歴（role, content, tool等を含む）
    tools : list[Tool]
        fastmcpで取得したToolオブジェクトのリスト
    model : str, optional
        利用するLLMモデル名（例: azure/gpt-4o-mini や azure/gpt-4.1-mini など、環境に合わせて）

    Returns
    -------
    ModelResponse
        LLMの応答
    """
    litellm_tools = fastmcp_tools_to_litellm_tools(tools)
    response = litellm.completion(
        model=model,
        messages=messages,
        tools=litellm_tools,
        tool_choice="auto",
        # 環境変数（AZURE_API_KEY / AZURE_API_BASE / AZURE_API_VERSION）を設定している場合は省略可
        api_base="<AOAIのAPIエンドポイント>",
        api_version="<APIバージョン>",
        api_key="<AOAIのAPIキー>",
    )
    return response  # type: ignore

async def request_completions(
    messages: list[dict],
    tools: list[Tool],
    client: Client,
    logger: logging.Logger,
):
    """
    チャット履歴とツール情報を使用してLLMからの応答を取得し、履歴を更新する。

    Notes
    -----
    ツールコールがある場合はMCPサーバのツールを実行し、結果を履歴に追加する。

    Parameters
    ----------
    messages : list[dict]
        チャット履歴（role, content）
    tools : list[Tool]
        MCPサーバのツール一覧
    client : Client
        MCPクライアント
    logger : logging.Logger
        ロガー

    Returns
    -------
    list[dict]
        追加されたメッセージ履歴（role, content）
    """
    history = messages.copy()
    logger.info(f"チャット履歴: {history}")
    while True:
        response = call_llm_with_tools(history, tools)
        logger.info("LLM（OpenAI）からの応答を受信: %s", response)
        message = response["choices"][0]["message"]
        # ツールコールがなければ終了
        if not (hasattr(message, "tool_calls") and message.tool_calls):
            history.append({"role": "assistant", "content": message["content"]})
            break

        # ツールコールがある場合
        history.append(
            {
                "role": "assistant",
                "content": f"ツールコールします: {message.tool_calls}",
                "tool_calls": message.tool_calls,
            }
        )
        # すべてのツールコールを実行
        for tool_call in message.tool_calls:
            tool_name = tool_call.function.name
            tool_args = tool_call.function.arguments
            logger.info(f"ツール呼び出し: {tool_name}({tool_args})")
            args_dict = json.loads(tool_args)
            logger.info(f"MCPツール {tool_name} を実行: 引数 {args_dict}")
            async with client:
                tool_result = await client.call_tool(tool_name, arguments=args_dict)
            logger.info(f"ツール {tool_name} の実行結果: {tool_result}")

            history.append(
                {
                    "tool_call_id": tool_call.id,
                    "role": "tool",
                    "name": tool_name,
                    "content": str(tool_result),
                }
            )
        # ツール結果をLLMに渡してループ継続
    return history