私たちは今、新たなソフトウェアの時代の入り口に立っています。Andrej KarpathyがYC Startup Schoolの基調講演で語ったSoftware 3.0、すなわち自然言語が主要なプログラミングインターフェースとなり、大規模言語モデルがコンピューティングエンジンであり自律的な協働者として機能する世界です。
Software 1.0（開発者が明示的な命令を書く）、Software 2.0（ニューラルネットワークがデータからパターンを学習する）とは異なり、Software 3.0は自然言語で動作します。私たちのプロンプトがプログラムとなり、LLMが私たちの意図を実行可能な行動に変換します。
このチュートリアルでは、AIエージェントが安全にコードを実行できるNovita AIサンドボックスへのリモートアクセスを提供する[MCPサーバー](https://github.com/Studio1HQ/mcp_remote_execution)を構築します。エージェント自体はmcp-useライブラリを使って作成され、MCPサーバーとの通信も自動的に管理されます。
## **MCP（モデルコンテキストプロトコル）とは？**
モデルコンテキストプロトコル（MCP）は、Anthropicが開発した、AIモデルが外部サービス、ツール、データソースと通信するためのオープンスタンダードです。「AI用のUSB-C」と考えてください。エージェントと外部世界の間の標準的なインターフェースを提供します。
### **3つのコアコンポーネント：**
1. **ツール:** サーバーによって公開される呼び出し可能な関数/API（例：ウェブブラウジング、コード実行）。
2. **リソース:** サーバーから提供される外部データで、通常はAIエージェントのコンテキストとして使用されます（例：ファイル、メタデータ、データセット）。
3. **プロンプト:** 上記のツールやリソースを操作する際のエージェントの行動をガイドするための詳細な指示。
MCP、そのアーキテクチャ、コンポーネントについて詳しくは、[最初のMCPサーバーを構築する](/build-your-first-mcp-server/)ブログをご覧ください。
### **AIツール統合におけるMCPの重要性**
MCPはAIエージェントと外部システムの双方向通信を確立します。これにより、LLMをカスタムツールに簡単に「接続」できるようになり、統合作業が減り、エラーの可能性が最小限に抑えられます。
### **mcp-useライブラリの紹介**
[mcp-use](https://github.com/mcp-use/mcp-use)ライブラリは、MCPサーバーとやり取りするAIエージェントの構築を簡素化するPythonパッケージです。エージェントの作成と、外部ツールやデータソースにアクセスするためのMCPサーバーとの通信管理を自動的に処理するため、アプリケーションロジックに集中できます。
## **Novita AI SandboxとModel APIの概要**
[**Novita AI Sandbox**](https://novita.ai/sandbox)
### **サンドボックスとは何か、なぜ重要か**
サンドボックスは、信頼できないコードをホストシステムに影響を与えずに実行できる、安全で隔離されたランタイム環境です。基本的には、AIエージェントがコードやコマンドを実行したり、ファイルを作成したりするための軽量な仮想コンピュータです。
Novita AIはこのサンドボックスをクラウドで提供しており、エージェントが必要に応じて迅速にアクセスでき、使用したリソースに基づく柔軟な秒単位の課金が可能です。
**Novitaサンドボックスの主な機能：**
* **安全な隔離:** 各サンドボックスには独立したファイルシステムと環境が割り当てられ、データを保護し、意図しない相互作用を防ぎます。
* **高速起動:** サンドボックスインスタンスは平均200ms未満で起動し、低レイテンシのシナリオに最適です。
* **マルチ言語対応:** Python、JavaScript、TypeScriptなど、複数のプログラミング言語でコードを実行できます。
* **素早い一時停止と再開:** サンドボックスをいつでも一時停止し、再開できます。ファイルシステムとプロセス状態は完全に復元されます。
* **バックグラウンド実行:** バックグラウンドタスクの実行をサポートし、結果を待つ必要があるシナリオに適しています。
### **Novita Model API：**
[**Novita AI Models**](https://novita.ai/models)
Novitaは、OpenAI、Google、DeepSeek、Qwenなどの主要な研究機関によるオープンソースAIモデルの広範なライブラリを提供しています。これらには、言語、視覚、音声、動画、埋め込み用のモデルが含まれます。当社の言語モデルはOpenAI SDKと完全に互換性があるため、OpenAIからNovitaに切り替えるには、クライアントのベースURLとAPIキーを更新し、Novitaモデルを選択するだけです。
||
|---|
|from openai import OpenAI<br /><br />client = OpenAI(<br />base\_url="https://api.novita.ai/v3/openai",<br />api\_key="<Your Novita API Key>",<br />)|
## **開発環境のセットアップ**
まず、GitHubリポジトリをクローンし、クリーンなPython環境をセットアップし、必要な依存関係をすべてインストールし、Novita AIのキーを取得します。
### **GitHubリポジトリをクローンし、uvで依存関係をインストールします。**
**1.** [uv](https://docs.astral.sh/uv/)（軽量なPythonパッケージマネージャー）をインストールします。
||
|---|
|pip install uv|
**2.** リポジトリ（[GitHub repo](https://github.com/Studio1HQ/mcp_remote_execution)）をクローンして、そのディレクトリに移動します。
||
|---|
|git clone https://github.com/Studio1HQ/mcp\_remote\_execution.git<br />cd mcp\_remote\_execution|
**3.** **uv**仮想環境を作成してアクティブ化します。
||
|---|
|\# 仮想環境を作成<br />uv venv<br /><br />\# 仮想環境をアクティブ化<br />source .venv/bin/activate # Mac/Linuxの場合<br />\# または<br />.venv\\Scripts\\activate # Windowsの場合|
**4.** プロジェクトの依存関係をインストールします。
||
|---|
|\# 依存関係をインストール<br />uv sync|
### **Novita AIアカウントを作成し、APIキーを取得する**
**1.** [novita.ai](https://novita.ai/user/register)でサインアップします。
**2.** ダッシュボードで、ユーザープロフィールアイコンにカーソルを合わせ、ポップアップで **API Keys** をクリックします。
**3.** Key Managementページで **Add New Key** をクリックします。ポップアップでキーの名前を入力し、**Confirm** をクリックして、生成されたキーをコピーします。
**4.** プロジェクトのディレクトリ内で、**.env** ファイルを作成し、以下を貼り付けます。
||
|---|
|NOVITA\_API\_KEY="<PASTE YOUR NOVITA API KEY HERE>"<br />NOVITA\_BASE\_URL="https://api.novita.ai/v3/openai"<br /><br />NOVITA\_E2B\_DOMAIN="sandbox.novita.ai"<br />NOVITA\_E2B\_TEMPLATE="code-interpreter-v1"|
### **Novita AIアカウントにクレジットを追加する。**
Novitaサンドボックスを使用するには、アカウントにクレジットを追加する必要があります。ダッシュボードタブで **'Billing'** をクリックします。次に、支払いページで支払い方法を追加し、少なくとも$10のクレジットをチャージします。
## **MCPサーバーとAIエージェントの統合を構築する**
環境が整ったので、構築を始めましょう。このプロセスには2つの部分があります。まず、MCPサーバーの作成を順を追って説明し、次にAIエージェントアプリケーション（MCPクライアントとして機能）を構築します。ただし、始める前に、MCPサーバーが使用する**サンドボックスマネージャー**を作成します。
### **サンドボックスマネージャー**
これは、サンドボックスインスタンスの起動と停止を担当します。今回のセットアップでは、サーバーを一度に1つのサンドボックスインスタンスに制限します。サンドボックスマネージャーは、サンドボックス内でのPythonコードの実行とシェルコマンドの実行も処理します。
まず、**sandbox\_manager.py** で、**SandboxManager** クラスを作成します。このクラスは以下のパラメータを受け取ります。
* **sandbox\_template**: サンドボックスインスタンスの作成に使用するテンプレート。今回は、一般的なPythonパッケージ（pandas、numpyなど）がプリインストールされている "code-interpreter-v1" を使用します。
* **sandbox\_domain**: Novita AIサンドボックスインスタンスに接続するためのドメインエンドポイント。
* **sandbox\_timeout**: サンドボックスが自動的に終了されるまでのアクティブ時間（秒）。
||
|---|
|from novita\_sandbox.code\_interpreter import Sandbox<br /><br /><br />class SandboxManager:<br />def \_\_init\_\_(<br />self,<br />sandbox\_template: str,<br />sandbox\_domain: str,<br />sandbox\_timeout: int,<br />):<br />self.sandbox\_template = sandbox\_template<br />self.sandbox\_domain = sandbox\_domain<br />self.sandbox\_timeout = sandbox\_timeout|
次に、サンドボックスを作成および停止するメソッドを追加します。
1. 作成では、**sandbox\_api\_key** を受け取り、それを使用して新しいサンドボックスインスタンスを起動し、成功メッセージとサンドボックスID、または何か問題が発生した場合の例外メッセージを返します。
2. 停止では、**sandbox\_api\_key** と **sandbox\_id** の両方を受け取り、サンドボックスに接続して（存在する場合）停止します。前と同様に、結果に応じて成功メッセージまたは例外メッセージを返します。
||
|---|
|...<br />class SandboxManager:<br />... # 上記の既存コードの下<br />def create\_sandbox\_session(self, sandbox\_api\_key: str) -> str:<br />"""<br />新しいサンドボックスインスタンスを作成します。<br /><br />Args:<br />sandbox\_api\_key (str): サンドボックスのAPIキー。<br /><br />Returns:<br />str: 新しいサンドボックスのサンドボックスIDを含む成功メッセージ、またはエラーメッセージ。<br />"""<br />try:<br /><br />\# 新しいサンドボックスを作成<br />sandbox = Sandbox.create(<br />template=self.sandbox\_template,<br />api\_key=sandbox\_api\_key,<br />domain=self.sandbox\_domain,<br />timeout=self.sandbox\_timeout,<br />)<br /><br />return f"Successfully created sandbox. Sandbox ID: {sandbox.sandbox\_id}"<br /><br />except Exception as e:<br />return f"Failed to create new sandbox:{str(e)}"<br /><br />def stop\_sandbox\_session(self, sandbox\_api\_key: str, sandbox\_id: str) -> str:<br />"""<br />サンドボックスインスタンスが存在する場合はそれを強制終了します。<br /><br />Args:<br />sandbox\_api\_key (str): サンドボックスのAPIキー。<br />sandbox\_id (str): サンドボックスのID。<br /><br />Returns:<br />str: 強制終了されたサンドボックスのIDを含む成功メッセージ、またはエラーメッセージ。<br />"""<br />try:<br />\# サンドボックスに接続<br />sandbox = Sandbox.connect(<br />api\_key=sandbox\_api\_key,<br />sandbox\_id=sandbox\_id,<br />)<br /><br />sandbox.kill()<br /><br />return f"Successfully killed Sandbox ID: {sandbox\_id}"<br /><br />except Exception as e:<br />return f"Failed to kill Sandbox ID: {sandbox\_id}\\ {str(e)}"|
最後に、APIキーとIDを介してユーザーのサンドボックスに接続した後、Pythonコードとシェルコマンドを実行するメソッドを追加します。すべてのサンドボックス出力（例外やエラーを含む）は辞書で返されます。
||
|---|
|...<br />class SandboxManager:<br />... # 上記の既存コードの下<br />def run\_python\_code(<br />self, python\_code: str, sandbox\_api\_key: str, sandbox\_id: str<br />) -> dict:<br />"""<br />サンドボックスでPythonコードを実行します。画像出力がある場合はスキップします。<br /><br />Args:<br />python\_code (str): 実行するPythonコード。<br />sandbox\_api\_key (str): サンドボックスのAPIキー。<br />sandbox\_id (str): サンドボックスのID。<br /><br />Returns:<br />dict: stdout、ログ、エラーなどを含む。<br />"""<br /><br />try:<br />\# サンドボックスに接続<br />sandbox = Sandbox.connect(<br />api\_key=sandbox\_api\_key,<br />sandbox\_id=sandbox\_id,<br />)<br /><br />execution = sandbox.run\_code(python\_code, language="python")<br /><br />return {<br />\# 画像出力はスキップ<br />"outputs": \[result for result in execution.results if not result.png\],<br />"logs": execution.logs,<br />"error": execution.error,<br />}<br /><br />except Exception as e:<br />return {"error": str(e)}<br /><br />def run\_on\_command\_line(<br />self, command: str, sandbox\_api\_key: str, sandbox\_id: str<br />) -> dict:<br />"""<br />サンドボックスでコマンドを実行します。<br /><br />Args:<br />command (str): 実行するコマンド。<br />sandbox\_api\_key (str): サンドボックスのAPIキー。<br />sandbox\_id (str): サンドボックスのID。<br /><br />Returns:<br />dict: コマンドの出力と実行エラー（あれば）を含む。<br />"""<br /><br />try:<br />\# サンドボックスに接続<br />sandbox = Sandbox.connect(<br />api\_key=sandbox\_api\_key,<br />sandbox\_id=sandbox\_id,<br />)<br /><br />result = sandbox.commands.run(command)<br />return {<br />"output": {<br />"stdout": result.stdout,<br />"stderr": result.stderr,<br />"exit\_code": result.exit\_code,<br />"error": result.error,<br />},<br />"execution error": None,<br />}<br /><br />except Exception as e:<br />return {"output": None, "execution error": str(e)}|
### **MCPサーバーの作成**
サンドボックスマネージャーの準備ができたので、**mcp\_server.py** に取り組みます。まず、サーバーを実行するフレームワークである **FastMCP** のインスタンスと、SandboxManager を作成します。また、ターミナルに美しい表示を行うためのRichコンソールインスタンスも作成します。
**注:** SandboxManager に渡されるサンドボックスタイムアウトは、このサーバーで起動されるすべてのサンドボックスに適用されます。これは、各サンドボックスが（早期に停止されない限り）生き続けることができる最大時間（秒）です。
||
|---|
|import asyncio<br />import os<br /><br />from dotenv import load\_dotenv<br />from mcp.server.fastmcp import Context, FastMCP<br />from rich.console import Console<br />from rich.panel import Panel<br />from rich.table import Table<br />from starlette.requests import Request<br /><br />from sandbox\_manager import SandboxManager<br /><br />\# .env 変数を読み込み<br />load\_dotenv()<br /><br />console = Console()<br /><br />\# FastMCPサーバーを初期化<br />mcp = FastMCP("MCP\_Server")<br /><br />\# シングルトンサンドボックスインスタンス用のサンドボックスマネージャーを初期化<br />sandbox\_manager = SandboxManager(<br />sandbox\_template=os.getenv("NOVITA\_E2B\_TEMPLATE"),<br />sandbox\_domain=os.getenv("NOVITA\_E2B\_DOMAIN"),<br />sandbox\_timeout=900, # 900秒（15分）。その後サンドボックスインスタンスは自動的に強制終了されます。<br />)<br />|
次に、ユーザーのサンドボックスに接続するにはAPIキーが必要なため、サーバーへのユーザーのリクエストからAPIキーを取得します。後でMCPクライアントで説明するように、このキーはすべてのリクエストのauthorizationヘッダーで送信されます。つまり、サーバー側でこのヘッダーからAPIキーを抽出する方法が必要です。
FastMCPは、実行時にリクエストを行っているユーザーに関する情報を保持する **Context** オブジェクトを提供します。そこで、コンテキストオブジェクトを受け取り、ヘッダーからAPIキーを抽出して返すヘルパーメソッド **get\_user\_api\_key** を作成します。キーがない場合は例外を発生させます。
||
|---|
|... # 上記の既存コードの下<br />def get\_user\_api\_key(ctx: Context) -> str:<br />"""<br />リクエストヘッダーからAPIキーを返します。存在しない場合は例外を発生させます。<br />"""<br /><br />request: Request = ctx.request\_context.request<br /><br />\# リクエストデータにアクセス<br />auth\_header = request.headers.get("Authorization")<br /><br />if auth\_header:<br />auth\_header = auth\_header.split(" ")\[1\]<br /><br />if not auth\_header:<br />raise Exception("Missing API Key in Authorization Bearer header")<br /><br />return auth\_header|
それでは、サーバー上でプロンプト、ツール、リソースを公開します。これを行うには、Python関数にデコレータ **@mcp.{prompt, tool, resource}()** を追加するだけです。最初にプロンプトを作成します。これは、エージェントがサンドボックスをどのように使用すべきかに関する指示を返します。
||
|---|
|... # 上記の既存コードの下（注: Console表示ヘルパーメソッドは簡潔にするため省略）<br /><br />@mcp.prompt()<br />def instructions\_for\_sandbox\_use() -> str:<br />"""<br />サンドボックス使用のための指示（必ず読むこと）を返します。<br />"""<br />return """<br />サンドボックス機能を使用する場合は、最初に create\_sandbox\_session() 関数を呼び出して新しいサンドボックスセッションを作成する必要があります。<br />その後、run\_python\_code() または run\_on\_command\_line() 関数を使ってサンドボックス上で実行できます。<br />使い終わったら、stop\_sandbox\_session() 関数を呼び出してサンドボックスセッションを強制終了する必要があります。<br /><br />注意:<br />\- サンドボックスには一般的なデータ分析パッケージがすでにプリインストールされていますが、パッケージが存在するかどうか確信が持てない場合や、不足しているパッケージが原因でコードにインポートエラーが発生した場合は、インストールされているか確認し、インストールされていなければインストールできます。<br />"""|
ツールについては、**sandbox\_manager** インスタンスを使用して、ユーザーのサンドボックスの作成、停止、コードまたはコマンドの実行を行うメソッドを公開します。
**注:** 以下のツールメソッドが受け取る引数のうち、**ctx: Context** を含めます。これにより、FastMCP にリクエストコンテキストをそのパラメータに自動的に挿入するよう指示します（これは依存性注入と呼ばれるプロセスです）。次に、このコンテキストをヘルパーメソッド（**get\_user\_api\_key**）に渡すことで、APIキーが存在すれば取得します。
||
|---|
|... # 上記の既存コードの下<br /><br />@mcp.tool()<br />def create\_sandbox\_session(ctx: Context) -> str:<br />"""<br />サンドボックスインスタンスを作成し、成功メッセージとサンドボックスID、またはエラーメッセージを返します。<br />"""<br />try:<br />return sandbox\_manager.create\_sandbox\_session(get\_user\_api\_key(ctx))<br />except Exception as e:<br />return e<br /><br /><br />@mcp.tool()<br />def stop\_sandbox\_session(sandbox\_id: str, ctx: Context) -> str:<br />"""<br />サンドボックスインスタンスが存在する場合、それを強制終了します。<br />"""<br />try:<br />return sandbox\_manager.stop\_sandbox\_session(get\_user\_api\_key(ctx), sandbox\_id)<br />except Exception as e:<br />return e<br /><br /><br />@mcp.tool()<br />def run\_python\_code(python\_code: str, sandbox\_id: str, ctx: Context) -> dict:<br />"""<br />サンドボックスでPythonコードを実行します。画像出力がある場合はスキップします。<br /><br />Args:<br />python\_code (str): 実行するPythonコード。<br />sandbox\_id (str): サンドボックスのID。<br /><br />注意:<br />ctx (Context) は自動的に渡される依存性注入オブジェクトです。<br /><br />Returns:<br />dict: stdout、ログ、エラーなどを含む。<br />"""<br />console.print(<br />Panel(<br />python\_code,<br />title="Agent Executing Python Code",<br />border\_style="blue",<br />)<br />)<br /><br />try:<br />result = sandbox\_manager.run\_python\_code(<br />python\_code, get\_user\_api\_key(ctx), sandbox\_id<br />)<br /><br />\# 結果を表示。注: 本番環境ではなくテストでのみ行ってください。<br />display\_sandbox\_code\_output(result)<br />return result<br /><br />except Exception as e:<br />return {"error": str(e)}<br /><br /><br />@mcp.tool()<br />def run\_on\_command\_line(command\_line: str, sandbox\_id: str, ctx: Context) -> dict:<br />"""<br />サンドボックスでコマンドを実行します。<br /><br />Args:<br />command\_line (str): 実行するコマンド。<br />sandbox\_id (str): サンドボックスのID。<br /><br />注意:<br />ctx (Context) は自動的に渡される依存性注入オブジェクトです。<br /><br />Returns:<br />dict: コマンドの出力と実行エラー（あれば）を含む。<br />"""<br />console.print(<br />Panel(<br />command\_line,<br />title="Agent Executing Command Line",<br />border\_style="blue",<br />)<br />)<br /><br />try:<br />result = sandbox\_manager.run\_on\_command\_line(<br />command\_line, get\_user\_api\_key(ctx), sandbox\_id<br />)<br /><br />\# 結果を表示。注: 本番環境ではなくテストでのみ行ってください。<br />display\_sandbox\_command\_output(result)<br />return result<br /><br />except Exception as e:<br />return {"execution error": str(e)}<br />|
次に、後で説明するデモシナリオの1つに使用する単一のリソースを公開します。このリソースは、ユーザーの株式ポートフォリオのモックを返します。プロンプトやツールとは異なり、リソースにはアクセス可能なURLを指定する必要があります。
||
|---|
|... # 上記の既存コードの下<br /><br />@mcp.resource("data://user\_stock\_portfolio")<br />def get\_user\_portfolio() -> dict:<br />"""<br />主要なインデックスETFと個別株にわたるユーザーのポートフォリオ保有を返します。<br /><br />Returns:<br />dict: ティッカーシンボル、数量、平均購入価格を含むポートフォリオ<br />"""<br />portfolio = {<br />"holdings": \[<br />\# 主要インデックスETF<br />{<br />"ticker": "SPY",<br />"name": "SPDR S&P 500 ETF",<br />"quantity": 4,<br />"avg\_purchase\_price": 670.13,<br />"asset\_type": "ETF",<br />},<br />... # 簡潔にするため省略<br />\]<br />}<br /><br />return portfolio|
最後に、サーバーを起動するコードを追加します。Richコンソールはターミナルに出力し、これはブロッキング操作であって妨害になるため、**stdio** MCPトランスポートは使用しません。代わりに、**streamable-http** MCPトランスポートを使用します（これはクライアントとサーバー間のリモート接続に最も適しているため、本番環境でも使用されるものです）。
||
|---|
|... # 上記の既存コードの下<br /><br />if \_\_name\_\_ == "\_\_main\_\_":<br />\# サーバーを実行<br />\# 注意: コンソールに出力するため、stdioではなくstreamable-httpをトランスポートプロトコルとして使用します。stdioだとブロックされてしまいます。<br />\# また、本番環境ではstdioではなくSSEまたはstreamable-httpを使用する必要があります。<br /><br />asyncio.run(mcp.run(transport="streamable-http"))|
ふう、ではMCPサーバーを起動して、実行中のURLを確認しましょう。このURLはAIエージェントを接続するために必要になります。ターミナルで以下のコマンドを実行してください。
||
|---|
|uv run mcp\_server.py|
サーバーが起動したはずです。実行中のURLをメモしておいてください。
### **AIエージェントをMCPサーバーと統合する**
次に、**mcp\_client.py** でAIエージェントの作業を開始します。**mcp-use** ライブラリを使用すると、このプロセスが簡単になります。まず、エージェントの動作を確認するためにデバッグレベルをINFOに設定します。mainメソッドで、mcpクライアントの設定辞書を作成します。これは、利用可能なMCPサーバーを指定します。名前（私は **"stock&sandbox"** を使用）と、MCPサーバーが実行されているURL（末尾に ***/mcp*** を追加することを忘れないでください）を指定し、さらにユーザーのAPIキーを **"auth"** の値として含めます。これは **mcp-use** によってすべてのリクエストのAuthorization Bearerヘッダーに自動的に挿入されます。
mcp-useはlangchain-openaiに依存しているため、NovitaのベースURL、APIキー、LLMモデル名を渡します。NovitaはOpenAI互換であるため、これで動作します。
また、エージェントの応答に加えて、サーバー上で使用されたサンドボックスのID（もしあれば）を含めたいと思います（理由は後で説明します）。そのために、応答形式を表すPydanticクラスを定義します。
||
|---|
|import asyncio<br />import os<br />from datetime import datetime<br />from typing import Optional<br /><br />import mcp\_use<br />from dotenv import load\_dotenv<br />from langchain\_openai import ChatOpenAI<br />from mcp\_use import MCPAgent, MCPClient<br />from pydantic import BaseModel, Field<br />from rich.console import Console<br />from rich.panel import Panel<br />from rich.prompt import Prompt<br /><br />\# 環境変数を読み込み<br />load\_dotenv()<br /><br />console = Console()<br /><br />\# 注意: 1はINFOレベル、2は詳細なDEBUGレベル、0はデバッグ出力なし。<br />mcp\_use.set\_debug(1)<br /><br /><br />class ResponseFormat(BaseModel):<br />response: str<br />id\_of\_used\_sandbox: Optional\[str\] = Field(<br />..., description="使用されたサンドボックスのID（あれば）"<br />)<br /><br /><br />async def main(model: str, base\_url: str, api\_key: str):<br /><br />\# 設定辞書を作成<br />config = {<br />"mcpServers": {<br />"stock&sandbox": {<br />\# MCPサーバーの実行URLが異なる場合は以下を変更してください。<br />\# また、末尾に /mcp を追加するのを忘れないでください。<br />"url": "http://127.0.0.1:8000/mcp",<br />"auth": api\_key,<br />}<br />}<br />}<br /><br />\# 設定辞書からMCPClientを作成<br />client = MCPClient(config)<br /><br />\# LLMを作成<br />llm = ChatOpenAI(model=model, base\_url=base\_url, api\_key=api\_key)<br />...|
次に、LLM、MCPクライアント、最大ステップ数（エージェントが応答する前に実行できるアクション数を制限）、およびメモリを有効にして（mcp-useが会話履歴を自動的に管理するように）、MCPエージェントを作成します。また、現在の日付と時刻を含むシステムプロンプトと、サンドボックス使用に関するカスタム指示も提供します（一部のモデルがサーバーで公開されている指示プロンプトを読み忘れるため、これを追加しました）。
最後に、標準的な会話ループをセットアップします。ユーザー入力を取得し、それをagent.run()に渡し、応答を出力します。サンドボックスが使用された場合は、sessionを作成してサーバー上の停止メソッドを手動で呼び出します。これは、モデルがサンドボックスを閉じるのを忘れた場合の安全策として行います。
||
|---|
|...<br />async def main(model: str, base\_url: str, api\_key: str):<br /><br />... # 上記の既存コードの下<br /><br />\# クライアントを使用してエージェントを作成<br />agent = MCPAgent(<br />llm=llm,<br />client=client,<br />max\_steps=25,<br />memory\_enabled=True, # mcp-useが会話履歴を自動的に管理<br />system\_prompt=f"""<br />あなたは役立つアシスタントです。現在の日付は {datetime.now().strftime('%Y-%m-%d')} です。<br /><br />必ず覚えておくこと:<br />\- ツールを呼び出す前に、まず instructions\_for\_sandbox\_use() を呼び出して内容を読んでください。<br />\- サンドボックスを使用した後は、ユーザーに応答する前に必ず stop\_sandbox\_session() を呼び出してください。<br />""",<br />)<br /><br />console.print(<br />Panel(<br />"\[bold green\]MCP Session Started\[/bold green\]\\ 終了するには 'quit()' と入力してください。",<br />title="MCP Session",<br />border\_style="green",<br />)<br />)<br /><br />while True:<br />user\_input = Prompt.ask("\\ \[bold yellow\]>>> User Message\[/bold yellow\]")<br /><br />if user\_input.lower().strip() == "quit()":<br />break<br /><br />\# クエリをエージェントに渡し、応答を待機<br />response\_obj = await agent.run(user\_input, output\_schema=ResponseFormat)<br /><br />console.print(<br />f"\\ \[bold green\]>>> Assistant Response: {response\_obj.response} \[/\]"<br />)<br /><br /><br />if response\_obj.id\_of\_used\_sandbox:<br />\# サンドボックスがまだアクティブな場合、MCPサーバー上で閉じる処理をトリガー<br />session = await client.create\_session("stock&sandbox")<br />await session.call\_tool(<br />name="stop\_sandbox\_session",<br />arguments={"sandbox\_id": response\_obj.id\_of\_used\_sandbox},<br />)<br />await session.disconnect()<br />|
最後に、クライアントを起動するコードを追加します。
||
|---|
|... # 上記の既存コードの下<br /><br />if \_\_name\_\_ == "\_\_main\_\_":<br />asyncio.run(<br />main(<br />model="qwen/qwen3-coder-480b-a35b-instruct",<br />base\_url=os.getenv("NOVITA\_BASE\_URL"),<br />api\_key=os.getenv("NOVITA\_API\_KEY"),<br />)<br />)<br />|
### **MCP AIエージェントのテスト実行**
ターミナルで以下のコマンドを実行して、AIエージェントアプリケーションを起動できます。
||
|---|
|uv run mcp\_client.py|
以下は、次のプロンプトに対するデモ実行のビデオリンクです。
* **ユーザープロンプト1:** *「私は2,000ドルを持っています。過去6ヶ月間の主要米国株価指数のパフォーマンスをyfinanceから取得し、MLモデルを実行して、今後2ヶ月間のリターンを最大化するためにこの投資をどのように配分するかを予測してください。」*（リンクを挿入）
* **ユーザープロンプト2:** *「米国経済のデフレ・クラッシュのシミュレーションを複数実行し、最も可能性の高いものを選び、それが私の株式ポートフォリオに与える影響を説明してください。」*（リンクを挿入）
## **本番環境のMCPサーバーに関するヒント**
このチュートリアルはNovita Sandboxで動作するMCPサーバーの構築に焦点を当てていますが、本番環境へのデプロイには追加の考慮事項が必要です。
**適切なトランスポートを使用する:** *stdio* はローカル開発には適していますが、本番環境のMCPサーバーは *streamable-http* を使用してリモート接続を可能にする必要があります（上記で行ったように）。
**認証を実装する:** 上記のように、mcpサーバーのエンドポイントを認証で保護してください。各クライアントが必要なツールとリソースにのみアクセスできるようにします。詳細な認証方法については、[FastMCP authentication](https://gofastmcp.com/servers/auth/authentication) をご覧ください。
**ロギングを有効にする:** ロガーを使用してサーバーアクティビティを監視し、問題をデバッグし、使用パターンを追跡します。これはメンテナンスとトラブルシューティングに不可欠です。
**レート制限とクォータ:** レート制限とクォータを実装して、サーバーを悪用から保護します。これはリソース集約型のツールを公開する場合に特に重要です。
**ドキュメントとバージョン管理:** MCPサーバーのAPIの明確なドキュメントとバージョン管理を維持して、開発者やLLMが統合しやすくします。
## **結論**
これで、AIエージェントが自然言語の指示を通じてリモートでコードを実行できるMCPサーバーを構築できるようになりました。これは、Software 3.0の実用的な実装です。
このチュートリアルでは、コード実行機能を持つMCPサーバーの構築方法、サンドボックスのライフサイクル管理方法、サーバーに接続するmcp-useを使用したAIエージェントの作成方法を学びました。次は、データベースアクセス、ウェブ検索、または複数のサーバーを1つのエージェントにリンクして拡張してみてください。[Novita](https://novita.ai/) にアクセスしてください。AIエージェントを構築するために必要なツールが揃っています。
> Novita AIは、開発者がシンプルなAPIを使ってAIモデルを簡単にデプロイできるAIクラウドプラットフォームであり、手頃な価格で信頼性の高いGPUクラウドも提供しています。
