ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-23 22:14:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Use PEP 649 annotations

Browse source
This commit is contained in:
Jeremiah Lowin 2025-04-26 07:56:36 -04:00
commit 3fb089a5b6
1 changed files with 14 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

26
src/fastmcp/server/server.py
View file

@ -1,5 +1,7 @@
"""FastMCP - A more ergonomic interface for MCP servers."""
from __future__ import annotations
import datetime
from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import (
@ -63,7 +65,7 @@ class MountedServer:
def __init__(
self,
prefix: str,
server: "FastMCP",
server: FastMCP,
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
@ -149,7 +151,7 @@ class TimedCache:
@asynccontextmanager
async def default_lifespan(server: "FastMCP") -> AsyncIterator[Any]:
async def default_lifespan(server: FastMCP) -> AsyncIterator[Any]:
"""Default lifespan context manager that does nothing.
Args:
@ -162,8 +164,8 @@ async def default_lifespan(server: "FastMCP") -> AsyncIterator[Any]:
def _lifespan_wrapper(
app: "FastMCP",
lifespan: Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]],
app: FastMCP,
lifespan: Callable[[FastMCP], AbstractAsyncContextManager[LifespanResultT]],
) -> Callable[
[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]
]:
@ -183,7 +185,7 @@ class FastMCP(Generic[LifespanResultT]):
instructions: str | None = None,
lifespan: (
Callable[
["FastMCP[LifespanResultT]"],
[FastMCP[LifespanResultT]],
AbstractAsyncContextManager[LifespanResultT],
]
| None
@ -277,7 +279,7 @@ class FastMCP(Generic[LifespanResultT]):
self._mcp_server.get_prompt()(self._mcp_get_prompt)
self._mcp_server.list_resource_templates()(self._mcp_list_resource_templates)
def get_context(self) -> "Context[ServerSession, LifespanResultT]":
def get_context(self) -> Context[ServerSession, LifespanResultT]:
"""
Returns a Context object. Note that the context will only be valid
during a request; outside a request, most methods will error.
@ -770,7 +772,7 @@ class FastMCP(Generic[LifespanResultT]):
def mount(
self,
prefix: str,
server: "FastMCP[LifespanResultT]",
server: FastMCP[LifespanResultT],
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
@ -795,7 +797,7 @@ class FastMCP(Generic[LifespanResultT]):
async def import_server(
self,
prefix: str,
server: "FastMCP[LifespanResultT]",
server: FastMCP[LifespanResultT],
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
@ -869,7 +871,7 @@ class FastMCP(Generic[LifespanResultT]):
@classmethod
def from_openapi(
cls, openapi_spec: dict[str, Any], client: httpx.AsyncClient, **settings: Any
) -> "FastMCPOpenAPI":
) -> FastMCPOpenAPI:
"""
Create a FastMCP server from an OpenAPI specification.
"""
@ -879,8 +881,8 @@ class FastMCP(Generic[LifespanResultT]):
@classmethod
def from_fastapi(
cls, app: "Any", name: str | None = None, **settings: Any
) -> "FastMCPOpenAPI":
cls, app: Any, name: str | None = None, **settings: Any
) -> FastMCPOpenAPI:
"""
Create a FastMCP server from a FastAPI application.
"""
@ -898,7 +900,7 @@ class FastMCP(Generic[LifespanResultT]):
)
@classmethod
def from_client(cls, client: "Client", **settings: Any) -> "FastMCPProxy":
def from_client(cls, client: Client, **settings: Any) -> FastMCPProxy:
"""
Create a FastMCP proxy server from a FastMCP client.
"""