mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-08 22:21:35 +00:00
89c87ae59a
Add an e2e suite at tests/e2e/mcp/ that proves MCP authorization over the api_key auth family. An admin registers an upstream MCP server through the management API (POST /v1/mcp/server, persisted in the DB and picked up without a restart) and queues its deletion. Two keys are created against that one server: one granted access through object_permission.mcp_servers and one with no MCP grant. The permitted key is a live control proving the upstream is reachable and the tool is callable, so a denial on the ungranted key is an authorization decision rather than a dead server. The denied key then sees none of the server's tools on tools/list and is refused a tools/call with a 403 access_denied. A deterministic self-hosted FastMCP upstream (add/multiply over streamable-http) is added to the e2e compose stack so the suite runs offline with a known tool set. KeyGenerateBody gains an optional typed object_permission so the shared gateway can create a key with an MCP grant.
40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
"""Deterministic upstream MCP server for the mcp e2e suite.
|
|
|
|
A tiny FastMCP server exposing `add` and `multiply` over streamable-http so the
|
|
suite has a self-hosted, offline upstream to register and exercise. DNS-rebinding
|
|
protection is turned off because the litellm container reaches this over the
|
|
compose network by service name (`mcp-upstream:8090`), not localhost, and the
|
|
stack is an isolated throwaway. Bind host/port come from MCP_HOST/MCP_PORT.
|
|
"""
|
|
|
|
import os
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
from mcp.server.transport_security import TransportSecuritySettings
|
|
|
|
mcp: FastMCP = FastMCP(
|
|
"e2e-math",
|
|
host=os.getenv("MCP_HOST", "0.0.0.0"),
|
|
port=int(os.getenv("MCP_PORT", "8090")),
|
|
transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def add(a: int, b: int) -> int:
|
|
"""Add two integers"""
|
|
return a + b
|
|
|
|
|
|
@mcp.tool()
|
|
def multiply(a: int, b: int) -> int:
|
|
"""Multiply two integers"""
|
|
return a * b
|
|
|
|
|
|
def main() -> None:
|
|
mcp.run(transport="streamable-http")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|