Contributing Templates¶

"""FastAgentic application entry point."""
import os
from fastagentic import App
from fastagentic.protocols import enable_mcp, enable_a2a
from fastagentic.adapters.base import BaseAdapter  # Or your framework's adapter

# Your framework imports
from my_framework import MyAgent

from models.inputs import MyInput
from models.outputs import MyOutput


# Create custom adapter if needed
class MyFrameworkAdapter(BaseAdapter):
    """Adapter for My Framework."""

    def __init__(self, agent):
        self.agent = agent

    async def invoke(self, input: dict, config: dict | None = None):
        return await self.agent.run(input)

    async def stream(self, input: dict, config: dict | None = None):
        async for event in self.agent.stream(input):
            yield self.map_event(event)


# Initialize app
app = App(
    title="{{ project_name }}",  # Template variable
    version="1.0.0",
    durable_store=os.getenv("DURABLE_STORE", "redis://localhost:6379"),
)

# Enable protocols
enable_mcp(app)
enable_a2a(app)


# Define agent
my_agent = MyAgent(...)


# Register endpoint
@app.agent_endpoint(
    path="/run",
    runnable=MyFrameworkAdapter(my_agent),
    input_model=MyInput,
    output_model=MyOutput,
    stream=True,
    durable=True,
    mcp_tool="my_agent",
    a2a_skill="my-agent",
)
async def run_agent(input: MyInput) -> MyOutput:
    """Run the agent."""
    ...


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# tests/test_agent.py
"""Agent tests."""
import pytest
from app import app
from models.inputs import MyInput


@pytest.mark.asyncio
async def test_agent_endpoint_exists():
    """Verify agent endpoint is registered."""
    routes = [r.path for r in app.routes]
    assert "/run" in routes


@pytest.mark.asyncio
async def test_agent_input_validation():
    """Test input validation."""
    from fastapi.testclient import TestClient
    client = TestClient(app)

    # Missing required field should fail
    response = client.post("/run", json={})
    assert response.status_code == 422


@pytest.mark.asyncio
async def test_agent_basic_run():
    """Test basic agent execution."""
    # Mock the LLM to avoid API calls in tests
    ...


# tests/test_contracts.py
"""Protocol contract tests."""
from fastagentic.testing import validate_mcp_schema, validate_a2a_card
from app import app


def test_mcp_schema():
    """MCP schema is valid."""
    errors = validate_mcp_schema(app)
    assert not errors


def test_a2a_card():
    """A2A Agent Card is valid."""
    errors = validate_a2a_card(app)
    assert not errors