Serves a fixed set of :class:Resource / :class:ResourceTemplate over MCP.
Source code in autogen/beta/mcp/resources.py
| def __init__(self, resources: Sequence[Resource], templates: Sequence[ResourceTemplate]) -> None:
self._resources = tuple(resources)
self._templates = tuple(templates)
self._by_uri = {r.uri: r for r in self._resources}
self._compiled = [(_compile_template(t.uri_template), t) for t in self._templates]
|
Source code in autogen/beta/mcp/resources.py
| def register(self, server: "Server") -> None:
provider = self
# ``mcp``'s low-level decorators are untyped; ignore the resulting noise.
@server.list_resources() # type: ignore[no-untyped-call, misc]
async def _list_resources() -> list[MCPResource]:
return [_to_mcp_resource(r) for r in provider._resources]
@server.read_resource() # type: ignore[no-untyped-call, misc]
async def _read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]:
return await provider.read(str(uri))
if self._templates:
@server.list_resource_templates() # type: ignore[no-untyped-call, misc]
async def _list_resource_templates() -> list[MCPResourceTemplate]:
return [_to_mcp_template(t) for t in provider._templates]
|
async
Source code in autogen/beta/mcp/resources.py
| async def read(self, uri: str) -> list[ReadResourceContents]:
resource = self._by_uri.get(uri)
if resource is not None:
data = await call_user_fn(resource.read)
return [ReadResourceContents(content=data, mime_type=resource.mime_type)]
for pattern, template in self._compiled:
match = pattern.match(uri)
if match is not None:
data = await call_user_fn(template.read, match.groupdict())
return [ReadResourceContents(content=data, mime_type=template.mime_type)]
raise MCPResourceNotFoundError(uri)
|