ローカルLLMとMCPサーバーを触ってみる

コーディングエージェントを探していたがコレ！というのが見つからないままJunieを使っています

それで困ってなかったんですけど、そろそろ新しい情報をインプットしたいぜーってことでやっとこさMCPを動かしてみようかと

とりあえずMCPを知る

そもそもAIってすごいんやろ？！レベルの知識量からすると、MCPってピンとこなかったんですよねー

LLMってのは思考力があるすごいやつ！くらいの認識だと、なんでMCPなんてプロトコルが要るんや……っていう。

でも実際はLLM自体に情報を集める力なんてないですし、何かを操作する力なんてないんですよね。

シナプスがしっかりしている脳みそ。ってなだけで、五感があるわけでもないし、手足が生えているわけでもない。

これまではアプリケーションの一部として五感も手足も用意していた。

オリジナルの五感を用意して脳みそに情報を渡して、脳みそが出してきた情報を何とか解釈して手足を動かしていた。

これがなかなか大変で、脳みそもその時々で五感の内容を読み取れたりと読み取れなかったりするし、

手足を動かすための形で返してくれることもあれば返してくれないこともあった。

この五感と脳みそ、脳みそと手足のやり取りを決まった形にしましょーや。ってのがMCPっていうプロトコル。

という認識やけど、例え方あってるかな？ちょっと不安

MCPのアーキテクチャ

例えばバックエンドで簡単なCRUDを構築しようとすると、DBにデータを入れたり、DBからデータを取ったりする必要がある。

だいたいは共通インターフェースに則ったドライバをどこかからとってきて、それを使ってDBと通信する。

そうすることで、アプリケーションからはPostgreSQLなのか、MySQLなのか、SQLiteなのかを意識せずにデータの操作ができる

でも実際は、PostgreSQLもMySQLもSQLiteも構造の違うデータがその向こうにある

SQL共通っぽいふりをしたそれぞれ独自解釈言語があって、独自のデータの持ち方があったりする

それぞれ違ったデータリソースから、データを取り出すデータベースサーバーがあって、

そのサーバーと通信するクライアントとしてドライバーがある。

この構造と似てて、データとしてファイルシステムやったり、それこそデータベースやったりが存在してて、

それぞれのデータにMCPサーバーがあって、MCPサーバーが各データを取ってきたり書き込んだりする。

MCPサーバーにはMCPクライアントを使って命令を送り、MCPクライアントはアプリケーションに組み込まれる

table:関係

データベース MCP

データファイル データソース

RDBMS MCPサーバー

SQL MCPプロトコル 正確には違うけど

クライアントドライバー MCPクライアント

って感じ？

USB type-c が例に上がってるけど、あっちはハードウェアやからイメージしづらかったり……

なんでクライアントとサーバーを分けるの？

例えば、コーディングエージェントを作りたい場合、

まずローカルファイルの操作は必要になる。

ディレクトリやファイルの読み込みなんてのはエージェントツールに組み込んでしまえっていいんじゃないか？と思う

ぱっと思いつく理由としては、クライアント、サーバーをそれぞれ好きな言語で実装するのが目的なんかなーと。

CLIツールやから軽量なインタプリタ言語でいいけど、サーバーでは複雑な処理をするからコンパイラ言語がいいなーみたいな

んで、そこで次の疑問が浮かぶ。

ってことは、MCPサーバーは大量に必要？

言語が違う場合はそうする必要があるし、言語が同じなら一つにまとめることもできる。というのが現実的な回答かな？

結局はpythonのサーバーコードはあって、それ以外にもいくつかの言語で提供される。みたいなのがイメージできる

もしかしたら勘違いしてるかも

LLMがツールを呼び出すと思ってない？

LLMに自然言語で命令を投げたら、このツールを呼び出してくれって命令がレスポンスに含まれていて、

それを実行するんじゃないか？

つまり、LLMに対して、アプリが呼び出せる外部機能を伝え、LLMがじゃあこの機能呼び出してよって言ってくるのかも

それならプロトコルってことで理解できる

これは、LLMに五感や手足を生やすのではないっぽいぞ

実際に触ってみないとわからない

まずはMPCサーバーを立てたい

が、この方法で立てればいいよ～ってのはぱっとは見つからない。

というのも、それぞれ独自にサーバーを用意していて、それを勝手に使って勝手に立ててねって感じ。

あるサーバーはnodejsで実装されていて、あるサーバーではpythonで実装されているみたいな状況。

このそれぞれのサーバーを自分で立ててねーってのは正直めんどくさい。

ってみんなが思ってるみたいね、いくつかのサーバーを管理したりまとめて使うためのツールが作られているみたい。

さらっと調べた範囲だと、LangChainとmcphostがでてきた。

LangChainは複数のLLMを扱うためのツールで、そこにMCP機能のためのアダプターが追加されたみたい。

LangChain自体を知らないので、今からやりたいことに対して大きすぎる気がしている。

mcphostはもう少し小さく、mpcにかかわる機能に限定されているきがするので、

まずはこっちでサーバーを立ててみようかと思う

stdio と sse

ちらっと調べただけでもサーバーの立て方が2種類あることがわかる

stdioは標準入出力をつかってデータのやり取りを行うみたい。

標準入出力を使うことから、ローカル通信であることは間違いないです。

それに対してsseはHTTPを使うので、ネットワークの壁をこえられる。

さらにサーバーからクライアントへの一方通行ではあるものの、ストリーム通信が可能。

sseの強みとしては、MCPをdockerで立てることができるところにあるみたい。

ファイルシステムを操作するようなケースはローカル内にサーバーが立ってないといけない気がするけど、

どこかのAPIをたたくだけなら、dockerで建てられたほうが環境構築が必要ないことから有利な気がする

あとはどっかの誰かがMCPをインターネットに公開してくれたら、使う側はクライアントだけ組み込めばいいってのも便利かも。

と思って調べてみたけど、まだそこまでは成熟してないっぽい

MCPサーバーをまとめて管理するツールで建てようかと思ったけど、これはちゃんと自前で立てておいたほうがよさそう

filesystemを借りてきて立てるのと、自分で簡単なサーバーを立ててみよう

Custom MCP Server

ひとまず簡単なMCPサーバーを立ててみる

uv を install

$ winget install --id=astral-sh.uv -e

プロジェクトを作成

$ uv init

$ uv venv

$ uv run ./main.py

JetbrainsのIDEを使っているなら、Package管理からvenvを追加したら向き先がそっちになるから便利

サーバーを作成

$ uv add fastmcp

$ touch server.py

server.pyの中身はfastmcpから借りてきた

code:server.py

# server.py

from fastmcp import FastMCP

mcp = FastMCP("Demo")

@mcp.tool()

def add(a: int, b: int) -> int:

"""Add two numbers"""

return a + b

if __name__ == "__main__":

mcp.run()

クライアントを作成

$ touch client.py

code:client.py

import asyncio

from fastmcp import Client

async def main():

# Connect via stdio to a local script

async with Client("server.py") as client:

tools = await client.list_tools()

print(f"Available tools: {tools}")

result = await client.call_tool("add", {"a": 5, "b": 3})

print(f"Result: {result}")

if __name__ == "__main__":

asyncio.run()

実行

$ uv run client.py

serverにresoruceを追加してみる

ここのサンプルにあるresourceを追加してみる

code:server.py

@mcp.resource("greeting://{name}")

def get_greeting(name: str) -> str:

"""Get a personalized greeting"""

return f"Hello, {name}!"

code:client.py

resources = await client.list_resource_templates()

print(f"Available resources: {resources}")

result = await client.read_resource("greeting://World")

print(f"Result: {result}")

Ollamaとどう組み合わせるか

まずはollamaのtoolsの動作を確認しておく

ollamaのインストール

ollamaのtoolsを使ってみる

$ uv add ollama

code:client.py

def add_two_numbers(a: int, b: int) -> int:

"""

Add two numbers

Args:

a: The first integer number

b: The second integer number

Returns:

int: The sum of the two numbers

"""

return a + b

if __name__ == "__main__":

functions = {

"add_two_numbers": add_two_numbers

}

response = ollama.chat(

model="qwen3",

)

print(response)

for tool in response.message.tool_calls or []:

function_to_call = functions.get(tool.function.name)

if function_to_call:

print('Function output:', function_to_call(**tool.function.arguments))

else:

print('Function not found:', tool.function.name)

$ uv run .\client.py

model='qwen3' created_at='2025-05-29T12:05:22.3877115Z' done=True done_reason='stop' total_duration=3468788600 load_duration=1691336400 prompt_eval_count=165 pro

Function output: 8

関数ならこんな感じでtoolsに簡単に入れられる

これをmpc連携にする必要がある

雑にclient.list_tools()の結果をollamaに渡してもエラーがでる

pydantic_core._pydantic_core.ValidationError: 1 validation error for Tool

pydanticのvalidationにひっかかる

mcp.types.Tool -> ollama.Tool

大体同じ形なんやけど、ちょっとだけ違うので丁寧にマッピングする

code:client.py

def to_ollama_tool(tool: mcp.types.Tool) -> dict:

properties = {}

for key, prop in tool.inputSchema.get("properties", {}).items():

# ollama.Toolのclassに沿ったdictを作る

return {

"type": "function",

"function": {

"name": tool.name,

"description": tool.description,

"parameters": {

"type": tool.inputSchema.get("type", "object"),

"properties": properties,

"required": tool.inputSchema.get("required", []),

},

},

}

async def main():

async with Client("server.py") as client:

tools = await client.list_tools()

response = ollama.chat(

model="qwen3",

tools=ollama_tools,

)

print("response", response)

for tool in response.message.tool_calls or []:

result = await client.call_tool(tool.function.name, tool.function.arguments)

if result:

print('Function output:', result)

else:

print('Function not found:', tool.function.name)

if __name__ == "__main__":

asyncio.run(main())

response model='qwen3' created_at='2025-05-29T16:36:54.8482451Z' done=True done_reason='stop' total_duration=1973622300 load_duration=10615500 prompt_eval_count=

Toolの変換が必要なものの、それ以外はそのままストレートに使えるかんじがある

ollamaのtoolsのjsonの形が公開されている

この形に合うようにmcpのtoolsから変形させてあげればOK

code:.json

{

"type": "function",

"function": {

"name": "add_two_numbers",

"description": "Add two numbers",

"parameters": {

"type": "object",

"required": [

"a",

"b"

],

"properties": {

"a": {

"type": "integer",

"description": "The first integer number"

},

"b": {

"type": "integer",

"description": "The second integer number"

}

}

}

}

}

tool_callsの結果を含めてollamaに渡し、最終結果を得る

code:client.py

async def main():

async with Client("server.py") as client:

tools = await client.list_tools()

response = ollama.chat(

model="qwen3",

messages=messages,

tools=ollama_tools,

)

print("response", response)

messages.append(response.message) # assistantのメッセージを追加

for tool_call in response.message.tool_calls or []:

print(f"Tool call: {tool_call}")

result = await client.call_tool(tool_call.function.name, tool_call.function.arguments)

if result:

print('Function output:', result)

messages.append({'role': 'tool', 'content': str(result)})

else:

print('Function not found:', tool_call.function.name)

response = ollama.chat(

model="qwen3",

messages=messages,

tools=ollama_tools,

)

print("response", response)

messages変数を作り、assistantの結果の追加とtoolの結果をtool roleで追加したってだけ

tool_callsがあればループするような作りにすれば、期待する結果が得られるんじゃないかなーという印象

なんちゃってfilesystemで動作確認

ファイルの読み込みをするためだけの簡単なMCPサーバーを作って、ファイルの中身の説明をさせてみる

code:server/filesystem.py

from fastmcp import FastMCP

import os

mcp = FastMCP("FileSystem")

@mcp.tool()

"""List files in a directory"""

try:

return os.listdir(path)

except OSError:

return []

@mcp.tool()

def cat(path: str) -> str:

"""Read a file"""

try:

with open(path, "r", encoding="utf-8") as f:

return f.read()

except FileNotFoundError:

return ""

ここでさっきまでと違うのは、lsでディレクトリの中身を見て、client.pyがあるかを探して読み込みを行うということを期待している

つまり、連続してtoolsの処理が行われることを期待している

なので、clientもtoolの呼び出しがおわるまでループする

code:client.py

async def main():

from server.filesystem import mcp

async with Client(mcp) as client:

tools = await client.list_tools()

response = ollama.chat(

model="qwen3",

messages=messages,

tools=ollama_tools,

)

print("response", response)

messages.append(response.message) # assistantのメッセージを追加

while True:

if not response.message.tool_calls:

break

for tool_call in response.message.tool_calls or []:

print(f"Tool call: {tool_call}")

result = await client.call_tool(tool_call.function.name, tool_call.function.arguments)

if result:

print('Function output:', result)

messages.append({'role': 'tool', 'content': str(result)})

else:

print('Function not found:', tool_call.function.name)

response = ollama.chat(

model="qwen3",

messages=messages,

tools=ollama_tools,

)

print("response", response)

あと、さっきまでと違って、ファイルパスをserverの直接を入力するのから、FastMCPサーバーを明示的に呼び出すように変えてる

calcとfilesystemを分けるため

ログが大量なので関連するところだけ抜粋

$ uv run client.py

1回目のmessage送信ではカレントディレクトリからlsしてねーというツール呼び出しがあり、ファイルの一覧を返している

Tool call: function=Function(name='ls', arguments={'path': '.'})

2回目のmessage送信ではclient.pyをcatしてねーというツール呼び出しがあり、ファイルの内容を返している

Tool call: function=Function(name='cat', arguments={'path': 'client.py'})

3回目のmessage送信でtool_callsがなくなり、レスポンスが返された

それっぽくなってきたんじゃない？！

TODO

streamable-http を試したい

複数のserverに対応するケースを試したい

更新履歴

2025/05/20