"""MCP Resources for exposing reference implementations.
Resources providing access to reference implementation code
for MCP servers, agents, and A2A protocols.
"""
import aiofiles
from fastmcp import FastMCP
from fastmcp.exceptions import NotFoundError
from ..config import get_config
from ..observability import get_tracer
router = FastMCP("implementations-resources")
tracer = get_tracer(__name__)
config = get_config()
# =============================================================================
# MCP Server Templates
# =============================================================================
@router.resource("templates://mcp/python")
async def mcp_template_python() -> str:
"""Python MCP server template - main server file.
A complete, standards-compliant Python MCP server template
using FastMCP framework.
Use this as a starting point for new Python MCP servers.
"""
with tracer.start_as_current_span("resource.mcp_template_python"):
return await _load_implementation(
"mcp-server-templates/python/src/mcp_server_template/server.py"
)
@router.resource("templates://mcp/python/config")
async def mcp_template_python_config() -> str:
"""Python MCP server template - configuration module.
Pydantic-based configuration with environment variable support.
"""
with tracer.start_as_current_span("resource.mcp_template_python_config"):
return await _load_implementation(
"mcp-server-templates/python/src/mcp_server_template/config.py"
)
@router.resource("templates://mcp/python/dockerfile")
async def mcp_template_python_dockerfile() -> str:
"""Python MCP server template - Dockerfile.
Multi-stage Docker build for production deployment.
"""
with tracer.start_as_current_span("resource.mcp_template_python_dockerfile"):
return await _load_implementation(
"mcp-server-templates/python/Dockerfile"
)
@router.resource("templates://mcp/python/pyproject")
async def mcp_template_python_pyproject() -> str:
"""Python MCP server template - pyproject.toml.
Project configuration with dependencies and build settings.
"""
with tracer.start_as_current_span("resource.mcp_template_python_pyproject"):
return await _load_implementation(
"mcp-server-templates/python/pyproject.toml"
)
@router.resource("templates://mcp/dotnet")
async def mcp_template_dotnet() -> str:
"""C#/.NET MCP server template - main program file.
A standards-compliant .NET MCP server template.
"""
with tracer.start_as_current_span("resource.mcp_template_dotnet"):
return await _load_implementation(
"mcp-server-templates/dotnet/src/McpServerTemplate/Program.cs"
)
# =============================================================================
# MCP Server Implementations
# =============================================================================
@router.resource("implementations://mcp/standards")
async def mcp_impl_standards() -> str:
"""Standards MCP server implementation - main server file.
Reference implementation of an MCP server providing
standards documentation and validation tools.
"""
with tracer.start_as_current_span("resource.mcp_impl_standards"):
return await _load_implementation(
"mcp-servers/standards/src/standards_mcp_server/server.py"
)
@router.resource("implementations://mcp/standards/validation")
async def mcp_impl_standards_validation() -> str:
"""Standards MCP server - validation tools implementation.
Example of MCP tool implementation for validation.
"""
with tracer.start_as_current_span("resource.mcp_impl_standards_validation"):
return await _load_implementation(
"mcp-servers/standards/src/standards_mcp_server/tools/validation.py"
)
@router.resource("implementations://mcp/standards/search")
async def mcp_impl_standards_search() -> str:
"""Standards MCP server - search tools implementation.
Example of MCP tool implementation for searching.
"""
with tracer.start_as_current_span("resource.mcp_impl_standards_search"):
return await _load_implementation(
"mcp-servers/standards/src/standards_mcp_server/tools/search.py"
)
@router.resource("implementations://mcp/standards/resources")
async def mcp_impl_standards_resources() -> str:
"""Standards MCP server - resources implementation.
Example of MCP resource implementation.
"""
with tracer.start_as_current_span("resource.mcp_impl_standards_resources"):
return await _load_implementation(
"mcp-servers/standards/src/standards_mcp_server/resources/standards.py"
)
@router.resource("implementations://mcp/healthcare")
async def mcp_impl_healthcare() -> str:
"""Healthcare MCP server implementation - README.
Domain-specific MCP server for healthcare/pharmacy operations.
"""
with tracer.start_as_current_span("resource.mcp_impl_healthcare"):
return await _load_implementation(
"mcp-servers/healthcare/README.md"
)
# =============================================================================
# Agent Implementations
# =============================================================================
@router.resource("implementations://agents/python/langchain")
async def agent_impl_langchain() -> str:
"""LangChain agent implementation - README.
Reference implementation of an agent using LangChain framework.
"""
with tracer.start_as_current_span("resource.agent_impl_langchain"):
return await _load_implementation(
"agents/python/langchain/README.md"
)
@router.resource("implementations://agents/python/langgraph")
async def agent_impl_langgraph() -> str:
"""LangGraph agent implementation - README.
Reference implementation of an agent using LangGraph framework.
"""
with tracer.start_as_current_span("resource.agent_impl_langgraph"):
return await _load_implementation(
"agents/python/langgraph/README.md"
)
@router.resource("implementations://agents/dotnet")
async def agent_impl_dotnet() -> str:
"""MS Agent Framework implementation - README.
Reference implementation using Microsoft Agent Framework.
"""
with tracer.start_as_current_span("resource.agent_impl_dotnet"):
return await _load_implementation(
"agents/dotnet/ms-agent-framework/README.md"
)
# =============================================================================
# A2A Implementations
# =============================================================================
@router.resource("implementations://a2a/python")
async def a2a_impl_python() -> str:
"""Python A2A implementation - README.
Reference implementation of A2A protocol in Python.
"""
with tracer.start_as_current_span("resource.a2a_impl_python"):
return await _load_implementation(
"a2a/python/README.md"
)
@router.resource("implementations://a2a/dotnet")
async def a2a_impl_dotnet() -> str:
"""C#/.NET A2A implementation - README.
Reference implementation of A2A protocol in .NET.
"""
with tracer.start_as_current_span("resource.a2a_impl_dotnet"):
return await _load_implementation(
"a2a/dotnet/README.md"
)
# =============================================================================
# Dynamic Resources - Parameterized access
# =============================================================================
@router.resource("templates://{category}/{name}/{file}")
async def get_template_file(category: str, name: str, file: str) -> str:
"""Get any template file by category, name, and file path.
Args:
category: Template category (mcp, agents, a2a)
name: Template name (python, dotnet)
file: File path within the template
Returns:
File content.
"""
with tracer.start_as_current_span("resource.get_template_file") as span:
span.set_attribute("template.category", category)
span.set_attribute("template.name", name)
span.set_attribute("template.file", file)
# Map category to folder
folder_map = {
"mcp": "mcp-server-templates",
"agents": "agents",
"a2a": "a2a",
}
folder = folder_map.get(category, category)
path = f"{folder}/{name}/{file}"
return await _load_implementation(path)
@router.resource("implementations://{category}/{name}/{file}")
async def get_implementation_file(category: str, name: str, file: str) -> str:
"""Get any implementation file by category, name, and file path.
Args:
category: Implementation category (mcp, agents, a2a)
name: Implementation name (standards, healthcare, etc.)
file: File path within the implementation
Returns:
File content.
"""
with tracer.start_as_current_span("resource.get_implementation_file") as span:
span.set_attribute("implementation.category", category)
span.set_attribute("implementation.name", name)
span.set_attribute("implementation.file", file)
# Map category to folder
folder_map = {
"mcp": "mcp-servers",
"agents": "agents",
"a2a": "a2a",
}
folder = folder_map.get(category, category)
path = f"{folder}/{name}/{file}"
return await _load_implementation(path)
# =============================================================================
# Helper Functions
# =============================================================================
async def _load_implementation(relative_path: str) -> str:
"""Load an implementation file from disk.
Args:
relative_path: Path relative to reference-implementations directory.
Returns:
File content.
Raises:
NotFoundError: If file doesn't exist.
"""
impl_root = config.get_reference_implementations_root()
file_path = impl_root / relative_path
if not file_path.exists():
raise NotFoundError(
f"Implementation file not found: {relative_path}"
)
async with aiofiles.open(file_path, mode='r', encoding='utf-8') as f:
return await f.read()
