Register tool names declared in the request as allowed. The server may delegate unknown tools to MCP if configured.
(
tools: Optional[List[Dict[str, Any]]], mcp_endpoint: Optional[str] = None
)
| 37 | |
| 38 | |
| 39 | def register_runtime_tools( |
| 40 | tools: Optional[List[Dict[str, Any]]], mcp_endpoint: Optional[str] = None |
| 41 | ) -> None: |
| 42 | """Register tool names declared in the request as allowed. |
| 43 | The server may delegate unknown tools to MCP if configured. |
| 44 | """ |
| 45 | # Reset per-request registry to avoid leakage across requests |
| 46 | global _runtime_mcp_endpoint |
| 47 | _ALLOWED_RUNTIME_TOOLS.clear() |
| 48 | _runtime_mcp_endpoint = None |
| 49 | if not tools: |
| 50 | return |
| 51 | try: |
| 52 | for t in tools: |
| 53 | name = None |
| 54 | fn = t.get("function") if "function" in t else t |
| 55 | if isinstance(fn, dict): |
| 56 | name = fn.get("name") or t.get("name") |
| 57 | else: |
| 58 | name = t.get("name") |
| 59 | if name: |
| 60 | _ALLOWED_RUNTIME_TOOLS.add(str(name)) |
| 61 | # Detect per-tool endpoint extension |
| 62 | ext_ep = ( |
| 63 | t.get("x-mcp-endpoint") |
| 64 | or t.get("x_mcp_endpoint") |
| 65 | or ( |
| 66 | isinstance(t.get("function"), dict) |
| 67 | and t["function"].get("x-mcp-endpoint") |
| 68 | ) |
| 69 | or None |
| 70 | ) |
| 71 | if ext_ep and not mcp_endpoint: |
| 72 | mcp_endpoint = ext_ep |
| 73 | # Capture per-request MCP endpoint if provided (explicit or via tool extension) |
| 74 | if mcp_endpoint: |
| 75 | _runtime_mcp_endpoint = mcp_endpoint |
| 76 | except Exception: |
| 77 | # be forgiving on malformed tools |
| 78 | pass |
| 79 | |
| 80 | |
| 81 | async def execute_tool_call(name: str, arguments_json: str) -> str: |