chacha/pyMCPBroker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

more informative tool explanations

Browse Source
This commit is contained in:
ChaCha
2026-04-14 00:43:32 +02:00
parent 500b37bd26
commit fd085cbe87
8 changed files with 174 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@@ -114,3 +114,8 @@ depending on what the mounted MCP source actually reports through `tools/list`.
## VS Code
The repository includes a minimal `.vscode/` setup for running and debugging pytest and for launching the broker with `smoke_config.json`.
## OpenAPI guidance for agents
The FastAPI/OpenAPI descriptions are intentionally directive. The broker hides business tools behind three stable entry tools, so agents should start with `meta_tree`, inspect with `meta_desc`, then execute with `meta_call`.

5
docs/SPEC.md
View File

@@ -233,3 +233,8 @@ Accepted options:
## Repository editor config
The repository may include a `.vscode/` directory with recommended Python extensions plus launch/task settings for pytest and for starting the broker against `smoke_config.json`. This editor config is optional and does not affect runtime behavior.
## Agent-facing OpenAPI behavior
The public OpenAPI descriptions are part of the protocol surface. They explicitly tell a model that business tools are hidden behind the broker tree, that `meta_tree` is the discovery entrypoint, that `meta_desc` returns the exact schema for a discovered tool path, and that `meta_call` must only be used with arguments matching that schema.

4
pyMCPBroker/__init__.py
View File

@@ -1,2 +1,2 @@
__all__ = ["__version__"]
__version__ = "0.2.1"
__all__ = ['__version__']
__version__ = '0.2.2'

80
pyMCPBroker/app.py
View File

@@ -10,6 +10,36 @@ from .broker import Broker, BrokerError
from .models import MetaCallRequest, MetaDescRequest, MetaTreeRequest
META_TREE_EXAMPLE = {
"basic": {
"summary": "Start at the root node",
"value": {"path": "/"},
}
}
META_DESC_EXAMPLE = {
"tool_path": {
"summary": "Inspect a discovered tool path",
"value": {"path": "/repo/read/get_file"},
}
}
META_CALL_EXAMPLE = {
"tool_call": {
"summary": "Call a discovered tool path",
"value": {
"path": "/repo/read/get_file",
"args": {
"owner": "myorg",
"repo": "demo-repo",
"ref": "main",
"filePath": "README.md",
},
},
}
}
def create_app(broker: Broker, shared_secret: str | None = None) -> FastAPI:
@asynccontextmanager
async def lifespan(_: FastAPI):
@@ -23,8 +53,11 @@ def create_app(broker: Broker, shared_secret: str | None = None) -> FastAPI:
title="pyMCPBroker",
version=__version__,
description=(
"Expose three stable meta-tools over MCP stdio sources. "
"Use meta_tree to navigate paths, meta_desc to inspect a path, and meta_call to execute a tool path."
"This broker hides business tools behind three stable entry tools. "
"The model should not expect the real business tools to appear directly in OpenAPI. "
"Instead, call meta_tree first to discover available paths, then call meta_desc on a chosen path to inspect it. "
"For a tool path, meta_desc returns the exact argument schema required before execution. "
"Finally, call meta_call to execute the hidden business tool behind that path."
),
lifespan=lifespan,
)
@@ -35,7 +68,10 @@ def create_app(broker: Broker, shared_secret: str | None = None) -> FastAPI:
auth = request.headers.get("authorization", "")
api_key = request.headers.get("x-api-key", "")
if auth != f"Bearer {shared_secret}" and api_key != shared_secret:
return JSONResponse({"ok": False, "error_code": "unauthorized", "message": "Missing or invalid secret"}, status_code=401)
return JSONResponse(
{"ok": False, "error_code": "unauthorized", "message": "Missing or invalid secret"},
status_code=401,
)
return await call_next(request)
@app.exception_handler(BrokerError)
@@ -47,28 +83,52 @@ def create_app(broker: Broker, shared_secret: str | None = None) -> FastAPI:
@app.get("/")
async def root() -> dict[str, object]:
return {"ok": True, "service": "pyMCPBroker", "version": __version__, "meta_tools": ["meta_tree", "meta_desc", "meta_call"]}
return {
"ok": True,
"service": "pyMCPBroker",
"version": __version__,
"meta_tools": ["meta_tree", "meta_desc", "meta_call"],
"usage_hint": "Start with meta_tree on '/' to discover hidden tool paths.",
}
@app.post(
"/meta_tree",
summary="meta_tree",
description="Meta-tool. Use this first to discover available paths. Call it on a node path to list its direct children.",
summary="Discover tool paths",
description=(
"Entry tool for discovery. Use this first when you want capabilities from this broker. "
"It lists the direct children of a node path in the broker tree. "
"The returned children may be more nodes for navigation or callable tool paths as leaves. "
"The real business tools are hidden behind this tree and are not exposed directly in OpenAPI. "
"Typical workflow: call meta_tree on '/', navigate until you find a useful tool path, then call meta_desc on that path."
),
openapi_extra={"requestBody": {"content": {"application/json": {"examples": META_TREE_EXAMPLE}}}},
)
async def meta_tree(request: MetaTreeRequest) -> dict[str, object]:
return broker.meta_tree(request.path)
@app.post(
"/meta_desc",
summary="meta_desc",
description="Meta-tool. Use this to inspect a path. For a tool path, it returns the exact argument schema to use before calling it.",
summary="Inspect a discovered path",
description=(
"Entry tool for inspection. Use this on any path previously discovered with meta_tree. "
"For a node path, it explains that node and may summarize its children. "
"For a tool path, it returns the exact argument schema required before execution. "
"Call this before meta_call unless you already know the required arguments exactly."
),
openapi_extra={"requestBody": {"content": {"application/json": {"examples": META_DESC_EXAMPLE}}}},
)
async def meta_desc(request: MetaDescRequest) -> dict[str, object]:
return broker.meta_desc(request.path)
@app.post(
"/meta_call",
summary="meta_call",
description="Meta-tool. Calls a tool path. The args object must match the schema previously returned by meta_desc.",
summary="Call a discovered tool path",
description=(
"Entry tool for execution. Use this only on a tool path previously discovered with meta_tree. "
"The args object must match the exact args_schema previously returned by meta_desc for that same path. "
"This is how the model invokes the hidden business tool behind the selected broker path."
),
openapi_extra={"requestBody": {"content": {"application/json": {"examples": META_CALL_EXAMPLE}}}},
)
async def meta_call(request: MetaCallRequest) -> dict[str, object]:
return broker.meta_call(request.path, request.args)

12
pyMCPBroker/broker.py
View File

@@ -106,7 +106,7 @@ class Broker:
"path": entry.path,
"type": "node",
"children": [{"path": child.path, "type": child.type, "summary": child.summary} for child in entry.children],
"usage_hint": "Use meta_desc on a leaf path before meta_call.",
"usage_hint": "This broker hides business tools behind paths. Start here, then inspect a useful child path with meta_desc.",
}
def meta_desc(self, path: str) -> dict[str, Any]:
@@ -118,7 +118,7 @@ class Broker:
"type": "node",
"summary": entry.summary,
"description": entry.description,
"usage_hint": "Use meta_tree to navigate child paths.",
"usage_hint": "This is a navigation node. Use meta_tree on this path to browse its direct children.",
}
if entry.children:
payload["children"] = [{"path": child.path, "type": child.type, "summary": child.summary} for child in entry.children]
@@ -132,7 +132,7 @@ class Broker:
"summary": entry.summary,
"description": entry.description,
"args_schema": args_schema,
"usage_hint": "Call meta_call with this path and args matching args_schema.",
"usage_hint": "Do not guess arguments. Copy this tool path exactly and call meta_call with args matching args_schema.",
}
example_args = entry.tool_meta.get("_broker_example_args")
if example_args is not None:
@@ -148,7 +148,11 @@ class Broker:
raise BrokerError(
"invalid_arguments",
exc.message,
{"path": entry.path, "required": list(schema.get("required", [])), "usage_hint": "Call meta_desc on the same path before retrying."},
{
"path": entry.path,
"required": list(schema.get("required", [])),
"usage_hint": "Your args do not match the required schema for this hidden tool. Re-read meta_desc for the same path and retry.",
},
) from exc
timeout = float(entry.tool_meta.get("_broker_timeout") or 30)
max_output_chars = entry.tool_meta.get("_broker_max_output_chars")

64
pyMCPBroker/models.py
View File

@@ -7,16 +7,42 @@ from pydantic import BaseModel, Field, model_validator
class MetaTreeRequest(BaseModel):
path: str = "/"
path: str = Field(
default='/',
description=(
"Node path to browse inside the broker tree. Use '/' first to discover the top-level entries. "
"This must be a node path, not a tool path."
),
examples=['/'],
)
class MetaDescRequest(BaseModel):
path: str
path: str = Field(
description=(
"Previously discovered broker path to inspect. This can be either a node path or a tool path. "
"Use this after meta_tree to understand what a path represents and, for a tool path, to get its exact argument schema."
),
examples=['/repo/read/get_file'],
)
class MetaCallRequest(BaseModel):
path: str
args: dict[str, Any] = Field(default_factory=dict)
path: str = Field(
description=(
"Callable tool path previously discovered with meta_tree and inspected with meta_desc. "
"This must be a leaf tool path, not a node path."
),
examples=['/repo/read/get_file'],
)
args: dict[str, Any] = Field(
default_factory=dict,
description=(
"Arguments for the selected tool path. These arguments must match the args_schema returned by meta_desc for the same path. "
"Do not invent fields that are not present in that schema."
),
examples=[{'owner': 'myorg', 'repo': 'demo-repo', 'ref': 'main', 'filePath': 'README.md'}],
)
class ToolOverride(BaseModel):
@@ -29,28 +55,28 @@ class ToolOverride(BaseModel):
class SourceConfig(BaseModel):
backend: Literal["stdio"]
backend: Literal['stdio']
command: str
tool_filter: list[str] = Field(default_factory=list)
tool_overrides: dict[str, ToolOverride] = Field(default_factory=dict)
path_aliases: dict[str, str] = Field(default_factory=dict)
@model_validator(mode="after")
def validate_aliases(self) -> "SourceConfig":
@model_validator(mode='after')
def validate_aliases(self) -> 'SourceConfig':
for tool_name, alias in self.path_aliases.items():
if not tool_name:
raise ValueError("path_aliases keys must not be empty")
if not alias or "/" in alias or alias in {".", ".."}:
raise ValueError(f"Invalid path alias for tool {tool_name!r}: {alias!r}")
raise ValueError('path_aliases keys must not be empty')
if not alias or '/' in alias or alias in {'.', '..'}:
raise ValueError(f'Invalid path alias for tool {tool_name!r}: {alias!r}')
return self
class NodeConfig(BaseModel):
path: str
type: Literal["node"] = "node"
summary: str = ""
description: str = ""
children: list["NodeConfig"] = Field(default_factory=list)
type: Literal['node'] = 'node'
summary: str = ''
description: str = ''
children: list['NodeConfig'] = Field(default_factory=list)
source: SourceConfig | None = None
@@ -64,9 +90,9 @@ class RootConfig(BaseModel):
@dataclass(slots=True)
class Entry:
path: str
type: Literal["node", "tool"]
summary: str = ""
description: str = ""
type: Literal['node', 'tool']
summary: str = ''
description: str = ''
@dataclass(slots=True)
@@ -77,6 +103,6 @@ class NodeEntry(Entry):
@dataclass(slots=True)
class ToolEntry(Entry):
backend_key: str = ""
tool_name: str = ""
backend_key: str = ''
tool_name: str = ''
tool_meta: dict[str, Any] = field(default_factory=dict)

2
pyproject.toml
View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "pyMCPBroker"
version = "0.2.1"
version = "0.2.2"
description = "Small FastAPI MCP broker exposing stable meta-tools over stdio MCP sources"
readme = "README.md"
requires-python = ">=3.10"

38
tests/test_openapi_guidance.py Normal file
View File

@@ -0,0 +1,38 @@
from __future__ import annotations
from fastapi.testclient import TestClient
from pyMCPBroker.app import create_app
from pyMCPBroker.broker import Broker
from pyMCPBroker.config import load_config
from test_broker_end_to_end import make_config
def test_openapi_descriptions_and_examples(tmp_path):
cfg = load_config(make_config(tmp_path))
broker = Broker(cfg)
app = create_app(broker)
with TestClient(app) as client:
schema = client.get('/openapi.json').json()
info_desc = schema['info']['description']
assert 'hides business tools' in info_desc
assert 'meta_tree first' in info_desc
tree_post = schema['paths']['/meta_tree']['post']
assert tree_post['summary'] == 'Discover tool paths'
assert 'not exposed directly in OpenAPI' in tree_post['description']
tree_examples = tree_post['requestBody']['content']['application/json']['examples']
assert tree_examples['basic']['value']['path'] == '/'
desc_post = schema['paths']['/meta_desc']['post']
assert desc_post['summary'] == 'Inspect a discovered path'
assert 'exact argument schema' in desc_post['description']
call_post = schema['paths']['/meta_call']['post']
assert call_post['summary'] == 'Call a discovered tool path'
assert 'hidden business tool' in call_post['description']
call_examples = call_post['requestBody']['content']['application/json']['examples']
assert call_examples['tool_call']['value']['path'] == '/repo/read/get_file'