feat: implement skeleton phase

This commit is contained in:
hh
2026-05-19 14:19:15 +02:00
parent 75a23d231e
commit 221e660c5c
21 changed files with 586 additions and 3 deletions
+13
View File
@@ -0,0 +1,13 @@
"""MCP server definitions and (later) the internal aggregator app."""
from __future__ import annotations
from beaver_gateway.mcp.types import (
HttpMcp,
McpServer,
McpServerT,
PythonToolMcp,
StdioMcp,
)
__all__ = ["HttpMcp", "McpServer", "McpServerT", "PythonToolMcp", "StdioMcp"]
+77
View File
@@ -0,0 +1,77 @@
"""User-facing MCP server declarations.
Three flavours, one factory facade (``McpServer``). The factory returns
discriminated-union members so downstream code can ``match`` on ``kind``.
"""
from __future__ import annotations
from collections.abc import Callable # noqa: TC003 — runtime use by pydantic
from typing import TYPE_CHECKING, Annotated, Literal
from pydantic import BaseModel, ConfigDict, Field
if TYPE_CHECKING:
from collections.abc import Iterable
class _BaseMcp(BaseModel):
model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
name: str
class StdioMcp(_BaseMcp):
"""Subprocess MCP server we spawn and connect to over stdio."""
kind: Literal["stdio"] = "stdio"
command: tuple[str, ...]
env: dict[str, str] | None = None
cwd: str | None = None
class HttpMcp(_BaseMcp):
"""Remote MCP server reached over streamable HTTP."""
kind: Literal["http"] = "http"
url: str
auth: str | None = None
class PythonToolMcp(_BaseMcp):
"""Bundle of Python callables exposed as one FastMCP namespace."""
kind: Literal["python_tool"] = "python_tool"
tools: tuple[Callable[..., object], ...]
McpServerT = Annotated[
StdioMcp | HttpMcp | PythonToolMcp, Field(discriminator="kind")
]
class McpServer:
"""Factory facade matching the PRD-documented config surface."""
@classmethod
def stdio(
cls,
*,
name: str,
command: Iterable[str],
env: dict[str, str] | None = None,
cwd: str | None = None,
) -> StdioMcp:
return StdioMcp(name=name, command=tuple(command), env=env, cwd=cwd)
@classmethod
def http(
cls, *, name: str, url: str, auth: str | None = None
) -> HttpMcp:
return HttpMcp(name=name, url=url, auth=auth)
@classmethod
def python_tool(
cls, *, name: str, tools: Iterable[Callable[..., object]]
) -> PythonToolMcp:
return PythonToolMcp(name=name, tools=tuple(tools))