# harness.yaml
mcp_servers:
- name: filesystem
command: "mcp-server-filesystem"
args: ["--root", "/workspace"]
transport: stdio
- name: postgres
command: "mcp-server-postgres"
args: ["--connection-string", "postgresql://..."]
transport: streamable_http
health_check:
interval: 30s
timeout: 5s# Harness初始化时
init_message = {
"jsonrpc": "2.0",
"id": 0,
"method": "initialize",
"params": {
"protocolVersion": "2025-11-25",
"capabilities": {
"sampling": True,
"roots": True,
"logging": True
}
}
}class ToolSchemaCache:
def __init__(self, cache_path="~/.harness/tool_cache.db"):
self.db = sqlite3.connect(cache_path)
self.db.execute("""
CREATE TABLE IF NOT EXISTS tool_schemas (
server_name TEXT,
tool_name TEXT,
schema TEXT,
schema_version TEXT,
cached_at INTEGER,
PRIMARY KEY (server_name, tool_name)
)
""")
def get_or_fetch(self, server: MCPServer, tool_name: str) -> dict:
# 1. 先查缓存
cached = self.db.execute(
"SELECT schema, schema_version FROM tool_schemas WHERE server_name = ? AND tool_name = ?",
(server.name, tool_name)
).fetchone()
if cached:
schema_json, cached_version = cached
# 2. 检查版本是否过期
current_version = server.get_schema_version(tool_name)
if current_version == cached_version:
return json.loads(schema_json) # 命中缓存
# 3. 缓存未命中,重新获取
schema = server.fetch_tool_schema(tool_name)
self.db.execute(
"INSERT OR REPLACE INTO tool_schemas VALUES (?, ?, ?, ?, ?)",
(server.name, tool_name, json.dumps(schema),
server.get_schema_version(tool_name), int(time.time()))
)
self.db.commit()
return schema# 连接池配置
HTTP_CLIENT = httpx.AsyncClient(
limits=httpx.Limits(max_connections=10, max_keepalive_connections=5),
timeout=30.0
)
async def stream_resource_from_server(server: MCPServer, uri: str) -> AsyncIterator[bytes]:
"""流式读取资源内容,避免一次性加载到内存"""
request = {
"jsonrpc": "2.0",
"id": generate_request_id(),
"method": "resources/read",
"params": {"uri": uri}
}
async with HTTP_CLIENT.stream("POST", server.endpoint, json=request) as response:
async for chunk in response.aiter_bytes(chunk_size=8192):
yield chunkclass ToolCallState(State):
"""工具调用状态"""
def __init__(self, server_name: str, tool_name: str, params: dict):
self.server = harness.get_mcp_server(server_name)
self.tool_name = tool_name
self.params = params
async def execute(self, context: ExecutionContext) -> StateTransition:
try:
# 1. 准备工具调用请求
request = {
"jsonrpc": "2.0",
"id": context.request_id,
"method": "tools/call",
"params": {
"name": self.tool_name,
"arguments": self.params
}
}
# 2. 发送请求(带超时和重试)
result = await self.server.call_with_retry(
request,
max_retries=3,
timeout=30
)
# 3. 检查结果
if "error" in result:
return StateTransition(
next_state="error_handling",
context={"error": result["error"]}
)
# 4. 结果写入上下文
context.set("tool_result", result["result"])
return StateTransition(next_state="next_step")
except TimeoutError:
return StateTransition(
next_state="retry_or_fallback",
context={"reason": "timeout"}
)class AtomicToolExecution:
"""原子性工具执行"""
def __init__(self, execution_id: str):
self.execution_id = execution_id
self.state_file = f"/tmp/harness/{execution_id}.state"
async def execute(self, server: MCPServer, request: dict) -> dict:
# 1. 检查是否已经执行过(幂等性)
if os.path.exists(self.state_file):
with open(self.state_file) as f:
cached_result = json.load(f)
return cached_result # 重复执行,直接返回缓存
# 2. 执行工具调用
result = await server.call(request)
# 3. 原子地写入状态(保证不会丢失)
os.makedirs(os.path.dirname(self.state_file), exist_ok=True)
with open(self.state_file + ".tmp", "w") as f:
json.dump(result, f)
os.rename(self.state_file + ".tmp", self.state_file) # 原子rename
return result# harness.yaml
permissions:
filesystem_server:
tools:
read_file:
allowed_paths: ["/workspace/**", "/tmp/**"]
denied_paths: ["/etc/**", "/home/**"]
write_file:
allowed_paths: ["/workspace/**"]
denied_paths: ["/**"]
postgres_server:
tools:
query:
allowed_tables: ["users", "orders"]
denied_tables: ["audit_log", "secrets"]
write:
allowed_operations: ["INSERT", "UPDATE"]
denied_operations: ["DROP", "DELETE"]async def call_tool(server_name: str, tool_name: str, params: dict) -> dict:
# 1. 检查权限策略
if not harness.policy_engine.is_allowed(server_name, tool_name, params):
raise PermissionDenied(f"Tool call denied by policy: {server_name}/{tool_name}")
# 2. 执行工具调用
return await harness.mcp_servers[server_name].call(tool_name, params)