Skip to content

DEVELOPMENT.md

Latest commit

 

History

History
79 lines (57 loc) · 2.29 KB

DEVELOPMENT.md

File metadata and controls

79 lines (57 loc) · 2.29 KB

DEVELOPER README

This document is for developers of sre-agent, specifically for v0.2.0.

To start the agent

Run the CLI once and complete the configuration wizard to create the user .env file in the platform config directory.

Start the agent server and the Slack MCP server:

docker compose up -d

Trigger an error on the store by adding loaf to the cart, or change the currency from EUR to GBP. (Note, there is an bug that errors might take some time to be indexed so if you trigger the agent immediately after you cause an error it might not be able to find the log.)

Trigger the locally running agent:

uv run python run.py /aws/containerinsights/no-loafers-for-you/application cartservice

Or:

uv run python run.py /aws/containerinsights/no-loafers-for-you/application currencyservice

Adding a New Tool

When adding a new tool/integration, follow one of these patterns:

Option 1: MCP Server

If an MCP server exists for the service, you can use that. No interface implementation is needed.

# tools/example.py
from pydantic_ai.mcp import MCPServerStdio
from sre_agent.core.settings import AgentSettings

def create_example_mcp_toolset(config: AgentSettings) -> MCPServerStdio:
    return MCPServerStdio(
        "docker",
        args=["run", "-i", "--rm", "-e", f"TOKEN={config.example.token}", "mcp/example"],
        timeout=30,
    )

Examples: github.py, slack.py

Option 2: Direct API

Use this when no MCP server is available. You must implement the relevant interface.

# tools/example.py
from sre_agent.interfaces import LoggingInterface
from sre_agent.models import LogQueryResult

class ExampleLogging(LoggingInterface):
    async def query_errors(
        self,
        source: str,
        service_name: str,
        time_range_minutes: int = 10,
    ) -> LogQueryResult:
        # Implementation using direct API calls
        ...

def create_example_toolset(config: AgentSettings) -> FunctionToolset:
    toolset = FunctionToolset()
    impl = ExampleLogging(config.example.api_key)

    @toolset.tool
    async def search_logs(...) -> LogQueryResult:
        return await impl.query_errors(...)

    return toolset

Examples: cloudwatch.py