MCPLite

valve

LLM アプリケーションにツール、リソース、プロンプトを公開するサーバーを構築するための、Model Context Protocol（MCP）の Python 実装です。

プロジェクトの目的

Valve は、MCP 準拠サーバーを作成し、LLM アプリケーションに統合するためのフレームワークです。プロトコルネゴシエーション、メッセージルーティング、トランスポート層の抽象化（stdio、direct、SSE）を処理し、複数の MCP サーバーを同時に管理するホストオーケストレーション層を提供します。このフレームワークにより、LLM は標準化されたプロトコルを介して外部のデータソースやツールを利用できます。

アーキテクチャの概要

• MCPLite：デコレーターを使用して MCP サーバーを定義する、サーバー側フレームワークです。機能を登録するための @tool、@resource、@prompt デコレーターを提供します。
• Host：複数の MCP クライアントを管理し、機能を集約してシステムプロンプトを生成し、LLM とのやり取りのためのエージェントループを実装するオーケストレーションエンジンです。
• Client：サーバーに接続し、機能のネゴシエーションを行ってリクエストを送信する MCP クライアントの実装です。
• Server：受信した MCP リクエストを登録済みのプリミティブにルーティングし、応答を返すメッセージプロセッサーです。
• Transport Layer：DirectTransport（プロセス内）、StdioTransport（stdio 経由のサブプロセス）、SSETransport（HTTP SSE）など、通信のための抽象化を提供します。
• Primitives：Python 関数を MCP メタデータでラップする、MCPTool、MCPResource、MCPResourceTemplate、MCPPrompt などの MCP の中核的な抽象化です。
• Messages：JSON-RPC 2.0 に準拠する、リクエスト、応答、通知、エラーなどの MCP プロトコルメッセージ用 Pydantic モデルです。
• Registry：ServerRegistry は実行可能なコードを含むサーバー側プリミティブを保持し、ClientRegistry は機能の検出に使用するクライアント側の定義を保持します。
• Inventory：利用可能な MCP サーバーをディレクトリからスキャンし、メタデータを維持するサーバーの検出・管理システムです。
• MCPChat：Chain フレームワーク上に構築されたチャットインターフェースで、会話型 LLM アプリケーションに MCP 機能を追加します。

依存関係

主な依存関係：

• pydantic：データの検証とシリアライズ
• requests：fetch サーバー用 HTTP クライアント
• beautifulsoup4：HTML の解析
• markdownify：HTML から Markdown への変換
• fastapi：SSE トランスポート用 Web フレームワーク
• sse-starlette：FastAPI 用 SSE サポート
• aiohttp：非同期 HTTP クライアント
• rich：ターミナルの書式設定と出力

ローカルの依存関係：

• Chain：Model、Message、MessageStore、Prompt、Chat クラスを提供する LLM フレームワーク（内部依存関係と思われます）

API ドキュメント

MCPLite

class MCPLite:
    def __init__(self, transport: Optional[Transport | str] = None)

MCP サーバーを作成するためのメインクラスです。

主なメソッド：

def tool(self, func: Callable) -> Callable

関数を MCP ツールとして登録するデコレーターです。関数には型アノテーションと docstring が必要です。

def resource(self, uri: str, mime_type: str = "text/plain", size: int = 1024) -> Callable

関数を MCP リソースまたはリソーステンプレートとして登録するデコレーターです。テンプレートの URI には {param} を使用します。

def prompt(self, func: Callable) -> Callable

関数を MCP プロンプトとして登録するデコレーターです。関数は文字列または PromptMessage オブジェクトのリストを返す必要があります。

def run(self)

サーバーを起動します。動作はトランスポートの種類によって異なります。

Host

class Host:
    def __init__(
        self,
        servers: list[str],
        model: str = "gpt",
        preferred_transport: transport_types = "stdio",
        console: Console = Console()
    )

複数の MCP サーバーを管理するオーケストレーションエンジンです。

パラメーター：

• servers：接続するサーバー名のリスト
• model：LLM のモデル識別子
• preferred_transport：優先するトランスポートの種類（"stdio"、"direct"、"sse"）
• console：出力用の Rich コンソール

主なメソッド：

def agent_query(self, prompt: str, message_store: MessageStore = MessageStore()) -> str | None

MCP ツール呼び出しを処理するエージェントループを実行します。最終回答または None を返します。

Client

class Client:
    def __init__(
        self,
        name: str = "Generic Client",
        transport: str | Transport | StdioClientTransport = "DirectTransport",
        server_function: Optional[Callable] = None
    )

サーバーに接続するための MCP クライアントです。

主なメソッド：

def initialize(self)

MCP ハンドシェイクと機能の検出を実行します。

def send_request(self, request: MCPRequest) -> MCPResult

サーバーにリクエストを送信し、結果を返します。

def send_notification(self, notification: MCPNotification)

サーバーに通知を送信します（応答は返されません）。

MCPChat

class MCPChat(Chat):
    def __init__(
        self,
        servers: list[str],
        model: str = "gpt",
        preferred_transport: transport_types = "stdio",
        **kwargs
    )

MCP 機能を備えたチャットインターフェースです。Chain の Chat クラスを継承しています。

パラメーター：

• servers：接続する MCP サーバー名のリスト
• model：モデル識別子
• preferred_transport：優先するトランスポートの種類

追加コマンド：

• /status：MCP 接続の状態を表示します
• /list_tools：利用可能なツールを一覧表示します
• /list_resources：利用可能なリソースを一覧表示します
• /list_prompts：利用可能なプロンプトを一覧表示します

トランスポートクラス

class StdioClientTransport(Transport):
    def __init__(self, server_command: list[str])

stdio 経由のサブプロセス通信に使用するクライアントトランスポートです。

class StdioServerTransport(Transport):
    def __init__(self)

stdio 通信に使用するサーバートランスポートです。

class DirectTransport(Transport):
    def __init__(self, server_function: Callable)

サーバー関数を直接呼び出すインプロセストランスポートです。

使用例

MCP サーバーの作成

from valve.mcplite.mcplite import MCPLite
from valve.transport import StdioServerTransport

mcp = MCPLite(transport=StdioServerTransport())

@mcp.tool
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

@mcp.resource(uri="myapp://status")
def get_status() -> str:
    """Get application status."""
    return "Server is running"

@mcp.prompt
def analysis_prompt(topic: str) -> str:
    """Generate an analysis prompt for a given topic."""
    return f"Please analyze the following topic in detail: {topic}"

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

Host を使った複数サーバーのオーケストレーション

from valve.host.Host import Host
from Chain import MessageStore

# Connect to multiple MCP servers
host = Host(
    model="gpt",
    servers=["fetch", "obsidian"],
    preferred_transport="stdio"
)

# Execute a query that may use multiple tools
message_store = MessageStore()
result = host.agent_query(
    "Fetch the content from example.com and save it to my notes",
    message_store
)
print(result)

MCP を使ったチャットアプリケーションの構築

from valve.mcpchat.mcpchat import MCPChat

# Create chat with MCP capabilities
chat = MCPChat(
    model="gpt",
    servers=["fetch", "obsidian"],
    preferred_transport="direct"
)

# Start interactive chat
chat.chat()

# Or use programmatically
response = chat.query("What's the weather in San Francisco?")