Skip to content

Development Guide

Development workflow, testing, and code quality commands.

Terminal window
# Install dependencies
uv sync --dev
# Activate environment
source .venv/bin/activate
# Run tests
pytest
# Format and lint
ruff format . && ruff check --fix .
# Type check
mypy src/
Terminal window
git checkout -b feature/your-feature-name

Edit code following project conventions (see Code Style section).

Terminal window
# Run all tests
pytest
# Run specific test file
pytest tests/test_prompts.py
# Run with coverage
pytest --cov=consumer_agent --cov-report=html
open htmlcov/index.html
Terminal window
# Auto-format code
ruff format .
# Check and fix lint issues
ruff check --fix .
# Sort imports
ruff check --select I --fix .
Terminal window
# Check all source files
mypy src/
# Check specific file
mypy src/consumer_agent/agent/agent.py
Terminal window
git add .
git commit -m "Brief description of changes
- Specific change 1
- Specific change 2
- Specific change 3"

Follow conventional commit format: feat:, fix:, docs:, refactor:, test:, chore:

Terminal window
git push origin feature/your-feature-name

Create PR on GitHub with description of changes.

Terminal window
# All tests (skips integration by default)
pytest
# Include integration tests
pytest -m integration
# Run with verbose output
pytest -v
# Run specific test
pytest tests/test_prompts.py::test_prompt_manager_loads_conversational_v3
# Run with debug output
pytest -v -s
Terminal window
# Generate coverage report
pytest --cov=consumer_agent --cov-report=html
# View in browser
open htmlcov/index.html
# Terminal report
pytest --cov=consumer_agent --cov-report=term

Create test files in tests/ with test_ prefix:

import pytest
from consumer_agent.agent import Agent
@pytest.mark.asyncio
async def test_agent_streaming():
"""Test agent streaming responses."""
model = create_chat_model()
agent = Agent(model)
messages = [{"role": "user", "content": "Hello"}]
events = []
async for event in agent.stream(messages, "You are helpful."):
events.append(event)
assert len(events) > 0

Mark integration tests:

@pytest.mark.integration
async def test_mcp_connection():
"""Test MCP server connection (requires server)."""
pass

Some tests make real OpenAI API calls and cost money. These are marked with @pytest.mark.requires_openai and are skipped by default.

Run expensive tests explicitly:

Terminal window
# Run all tests that require OpenAI (WARNING: costs money)
pytest -m requires_openai
# Run specific expensive test
pytest -m requires_openai tests/integration/test_landing_page_evaluation.py
# Run expensive integration tests only
pytest -m "requires_openai and integration"

Available test markers:

  • unit: Unit tests (no external dependencies)
  • integration: Integration tests (may require external services)
  • requires_openai: Tests that make OpenAI API calls (costs money)
  • requires_mcp: Tests that require MCP server
  • requires_aws: Tests that require real AWS credentials

Configuration in pyproject.toml:

[tool.ruff]
line-length = 100
target-version = "py311"
[tool.ruff.lint]
select = ["E", "F", "I", "N", "W", "UP"]
ignore = ["E501"]

Commands:

Terminal window
# Format code
ruff format .
# Check linting
ruff check .
# Fix auto-fixable issues
ruff check --fix .
# Check specific file
ruff check src/consumer_agent/agent/agent.py

Configuration in pyproject.toml:

[tool.mypy]
python_version = "3.11"
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true

Commands:

Terminal window
# Check all source files
mypy src/
# Check specific file
mypy src/consumer_agent/agent/agent.py
# Show error context
mypy --show-error-context src/

Install hooks:

Terminal window
pre-commit install

Run manually:

Terminal window
# All files
pre-commit run --all-files
# Staged files only
pre-commit run

Hooks run automatically on git commit.

  • Line length: 100 characters
  • Use type hints for all functions
  • Docstrings for public functions (Google style)
  • Import order: stdlib, third-party, local
# Good
def create_agent(model: BaseLanguageModel, tools: list[BaseTool] | None = None) -> Agent:
"""Create agent with tools."""
pass
# Bad - missing type hints
def create_agent(model, tools=None):
pass

Use Google style:

def load_config(config_path: str | None = None) -> dict[str, Any]:
"""Load agent configuration from YAML file.
Args:
config_path: Path to config file. If None, uses default.
Returns:
Configuration dictionary
Raises:
FileNotFoundError: If config file not found
yaml.YAMLError: If config file invalid
"""
pass

Order:

  1. Standard library
  2. Third-party packages
  3. Local imports
# Standard library
import os
from pathlib import Path
# Third-party
import yaml
from langchain.agents import create_agent
# Local
from consumer_agent.config import get_openai_config
Terminal window
# Development mode (auto-reload)
uvicorn consumer_agent.api.main:app --reload --port 8000
# Production mode
uvicorn consumer_agent.api.main:app --host 0.0.0.0 --port 8000

Test endpoints:

Terminal window
# Health check
curl http://localhost:8000/health
# Stream response
curl -X POST http://localhost:8000/agent/stream \
-H "Content-Type: application/json" \
-d '{
"messages": [{"role": "user", "content": "Hello"}],
"agent_id": "conversational",
"enabled_components": []
}'
Terminal window
# Via CLI command
consumer-agent chat
# Direct Python
python src/consumer_agent/cli/chat.py

Set log level in code:

from loguru import logger
import sys
logger.remove()
logger.add(sys.stderr, level="DEBUG")
# Install ipython
uv add --dev ipython
# Start interactive session
ipython
# In IPython:
from consumer_agent.agent import Agent
from consumer_agent.factory import create_chat_model
model = create_chat_model()
agent = Agent(model)
# During streaming
async for event in agent.stream(messages, system_prompt):
print(f"[DEBUG] Event: {event.event_type}, Content: {event}")
Terminal window
# Show print statements
pytest -s
# Drop into debugger on failure
pytest --pdb
# Drop into debugger at start
pytest --trace
  1. Create tool class in src/consumer_agent/utils/tools.py:
class NewTool(BaseTool):
name: str = "new_tool"
description: str = "Tool description"
args_schema: type[BaseModel] = NewToolInput
mcp_client: MCPClient
async def _arun(self, arg1: str, ...) -> str:
result = await self.mcp_client.call_tool("new_tool", {"arg1": arg1})
return str(result)
  1. Add to build_mcp_tools():
def build_mcp_tools(...) -> list[BaseTool]:
tools = [
# Existing tools...
NewTool(mcp_client),
]
return tools
  1. Test:
@pytest.mark.asyncio
async def test_new_tool():
# Test implementation
pass
  1. Create file in prompts/:
Terminal window
touch prompts/specialized-task-v1.txt
  1. Add content to file

  2. Load in code:

from consumer_agent.prompts import PromptManager
prompt = PromptManager.load_prompt("specialized-task-v1")
  1. Add test:
def test_specialized_task_prompt_loads():
prompt = PromptManager.load_prompt("specialized-task-v1")
assert len(prompt) > 0
Terminal window
# Add new dependency
uv add package-name
# Add dev dependency
uv add --dev package-name
# Update all dependencies
uv sync
# Update specific package
uv add package-name@latest
Terminal window
# Build wheel and sdist
uv build
# Output in dist/
ls dist/

Tests run automatically on:

  • Push to any branch
  • Pull request creation
  • Pull request updates

Check status:

Terminal window
# View recent workflow runs
gh run list
# View specific run
gh run view <run-id>
Terminal window
# Run full test suite
pytest --cov=consumer_agent
# Run linting
ruff check .
# Run type checking
mypy src/
# Run all checks (like CI)
ruff format . && ruff check --fix . && mypy src/ && pytest
consumer-agent/
├── src/consumer_agent/
│ ├── agent/ # Agent implementation
│ │ ├── agent.py # Main agent class
│ │ ├── prompts.py # PromptManager for system prompts
│ │ └── streaming.py # Event types
│ ├── api/ # FastAPI service
│ │ └── main.py # FastAPI application
│ ├── cli/ # CLI interface
│ │ ├── main.py # Click CLI
│ │ └── chat.py # TUI chatbot
│ ├── evaluation/ # Opik integration + metrics
│ ├── history/ # DynamoDB + S3 persistence
│ ├── utils/ # Utilities
│ │ ├── mcp_client.py # MCP JSON-RPC client
│ │ └── tools.py # MCP tool wrappers
│ ├── config.py # Configuration management
│ └── factory.py # Model + agent factory
├── prompts/ # System prompts
│ └── components/ # Feature-flagged YAML instructions
├── tests/ # Test suite
├── docs/ # Documentation
├── .env.example # Environment template
├── agent_config.yaml # Agent configuration
├── settings.yaml # Infrastructure configuration
└── pyproject.toml # Project metadata
Terminal window
# Environment
uv sync # Install deps
uv sync --dev # Install with dev deps
source .venv/bin/activate # Activate venv
# Testing
pytest # Run tests
pytest -v # Verbose
pytest --cov # With coverage
pytest -k test_name # Run specific test
# Code Quality
ruff format . # Format code
ruff check . # Lint code
ruff check --fix . # Fix lint issues
mypy src/ # Type check
# Running
uvicorn consumer_agent.api.main:app --reload # FastAPI
consumer-agent chat # TUI
python -m consumer_agent.cli.chat # TUI direct
# Dependencies
uv add package-name # Add dependency
uv add --dev package-name # Add dev dependency
uv remove package-name # Remove dependency
# Git
git status # Check status
git add . # Stage changes
git commit -m "message" # Commit
git push # Push to remote