|
1 |
| -For gpt expected container tool, here's an incomplete example |
2 |
| -Note that the SweRexManager or swe_rex are dummies, you need to implement your own container tool with session management |
| 1 | +# Container MCP Server Example |
| 2 | + |
| 3 | +This is an incomplete example showing how to implement a container tool for GPT using MCP. |
| 4 | + |
3 | 5 | ```
|
4 | 6 | from mcp.server.fastmcp import fastmcp
|
5 | 7 | # dummy showing how to import container tool
|
@@ -71,3 +73,72 @@ async def cleanup_session(ctx: Context) -> None:
|
71 | 73 | logger.info(f"Cleaning up session: {session_id}")
|
72 | 74 | await swe_rex_manager.cleanup_session(session_id)
|
73 | 75 | ```
|
| 76 | + |
| 77 | +### SweRexManager Implementation Pattern |
| 78 | + |
| 79 | +Based on the RemoteRuntime pattern, your SweRexManager could be implemented like below |
| 80 | +Note that this is a dummy implementation and you should implement your own version. |
| 81 | +``` |
| 82 | +from typing import Dict, Any, Optional |
| 83 | +import asyncio |
| 84 | +from swerex.runtime.remote import RemoteRuntime |
| 85 | +from swerex.runtime.config import RemoteRuntimeConfig |
| 86 | +
|
| 87 | +class SweRexManager: |
| 88 | + def __init__(self, config: Dict[str, Any]): |
| 89 | + """Initialize SweRexManager with dict configuration. |
| 90 | +
|
| 91 | + Args: |
| 92 | + config: Dictionary containing: |
| 93 | + - host: Server host (required) |
| 94 | + - port: Server port (optional) |
| 95 | + - timeout: Request timeout in seconds (optional, default 30.0) |
| 96 | + - auth_token: Authentication token (optional) |
| 97 | + """ |
| 98 | + self.config = RemoteRuntimeConfig(**config) |
| 99 | + self.runtime = RemoteRuntime.from_config(self.config) |
| 100 | + self.sessions: Dict[str, str] = {} # session_id -> runtime_session mapping |
| 101 | +
|
| 102 | + async def execute_in_session( |
| 103 | + self, |
| 104 | + session_id: str, |
| 105 | + cmd: list[str], |
| 106 | + workdir: Optional[str] = None, |
| 107 | + env: Optional[Dict[str, str]] = None, |
| 108 | + execution_timeout: int = 360, |
| 109 | + **kwargs |
| 110 | + ) -> str: |
| 111 | + """Execute command in a session.""" |
| 112 | + # Ensure session exists |
| 113 | + if session_id not in self.sessions: |
| 114 | + await self.create_session(session_id) |
| 115 | +
|
| 116 | + from swerex.runtime.abstract import Command |
| 117 | +
|
| 118 | + command = Command( |
| 119 | + command=cmd, |
| 120 | + timeout=execution_timeout, |
| 121 | + cwd=workdir, |
| 122 | + env=env or {} |
| 123 | + ) |
| 124 | +
|
| 125 | + response = await self.runtime.execute(command) |
| 126 | + return response.stdout if response.exit_code == 0 else response.stderr |
| 127 | +
|
| 128 | + async def create_session(self, session_id: str) -> None: |
| 129 | + """Create a new session.""" |
| 130 | + from swerex.runtime.abstract import CreateSessionRequest |
| 131 | +
|
| 132 | + request = CreateSessionRequest(session_id=session_id) |
| 133 | + await self.runtime.create_session(request) |
| 134 | + self.sessions[session_id] = session_id |
| 135 | +
|
| 136 | + async def cleanup_session(self, session_id: str) -> None: |
| 137 | + """Cleanup a session.""" |
| 138 | + if session_id in self.sessions: |
| 139 | + from swerex.runtime.abstract import CloseSessionRequest |
| 140 | +
|
| 141 | + request = CloseSessionRequest(session_id=session_id) |
| 142 | + await self.runtime.close_session(request) |
| 143 | + del self.sessions[session_id] |
| 144 | +``` |
0 commit comments