Skip to content

Contributing

Thank you for your interest in contributing to Adaptive Sentience!

Getting Started

Prerequisites

  • Python 3.9+
  • Git
  • Familiarity with asyncio and FastAPI
  • Understanding of distributed systems concepts

Setup Development Environment

# Clone repository
git clone https://github.com/adaptivesentience/agent_mesh.git
cd agent_mesh

# Create virtual environment
python3 -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

# Install development dependencies
pip install -r requirements-dev.txt

Run Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=. --cov-report=html

# Run specific test file
pytest tests/test_gateway.py

# Run with verbose output
pytest -v

Development Workflow

1. Create a Branch

# Create feature branch
git checkout -b feature/your-feature-name

# Or bug fix branch
git checkout -b fix/your-bug-fix

2. Make Changes

Follow the Code Style Guide below.

3. Test Your Changes

# Run tests
pytest

# Run linter
flake8 .

# Run type checker
mypy .

4. Commit Your Changes

# Stage changes
git add .

# Commit with descriptive message
git commit -m "Add feature: description of feature"

Commit message format: - Add feature: ... - New feature - Fix bug: ... - Bug fix - Refactor: ... - Code refactoring - Docs: ... - Documentation updates - Test: ... - Test additions/changes

5. Push and Create Pull Request

# Push to your fork
git push origin feature/your-feature-name

# Create pull request on GitHub
# Fill in PR template with description

Code Style

Python Style Guide

Follow PEP 8 with these additions:

Imports: 

# Standard library imports
import os
import sys
from typing import Dict, List, Optional

# Third-party imports
import fastapi
from pydantic import BaseModel

# Local imports
from gateway.router import Router
from trust.store import TrustStore

Type Hints: 

# Always use type hints
def execute_tool(tool_name: str, args: Dict[str, Any]) -> ToolResult:
    pass

# Use Optional for nullable values
def get_node(node_id: str) -> Optional[Node]:
    pass

Docstrings: 

def calculate_score(node: Node, tool: str) -> float:
    """Calculate selection score for a node.

    Args:
        node: Node to score
        tool: Tool name to check

    Returns:
        Score from 0.0 to 1.0

    Raises:
        ValueError: If node is invalid
    """
    pass

Async/Await: 

# Use async/await for I/O operations
async def fetch_data(url: str) -> Dict[str, Any]:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

Code Formatting

# Format with black
black .

# Sort imports
isort .

# Lint
flake8 .

Project Structure

agent_mesh/
├── gateway/              # HTTP gateway and routing
│   ├── http_gateway.py
│   ├── router.py
│   └── mesh_transport/
├── edge_node/            # Edge node runtime
│   ├── node.py
│   └── tools/
├── trust/                # Trust store and pairing
│   ├── store.py
│   └── pairing.py
├── runtime/              # Workflow execution
│   ├── workflow_runner.py
│   └── policies/
├── planner/              # Workflow planning
│   ├── planner.py
│   └── orchestrator/
├── discovery/            # Node discovery
│   ├── udp_discovery.py
│   └── mesh_snapshot.py
├── tests/                # Test suite
│   ├── test_gateway.py
│   └── test_trust.py
└── docs/                 # Documentation
    └── ...

Adding New Features

Adding a New Tool

  1. Create tool file:
# edge_node/tools/my_tool.py

from typing import Dict, Any
from pydantic import BaseModel

class MyToolInput(BaseModel):
    input_field: str

class MyToolOutput(BaseModel):
    output_field: str

async def execute(input: MyToolInput) -> MyToolOutput:
    """Tool implementation."""
    result = input.input_field.upper()
    return MyToolOutput(output_field=result)

# Tool metadata
TOOL_CONTRACT = {
    "tool_id": "tool:my_tool",
    "name": "my_tool",
    "version": "1.0.0",
    "description": "Description of what the tool does",
    "input_schema": MyToolInput.schema(),
    "output_schema": MyToolOutput.schema(),
    "required_capabilities": ["tool:my_tool"],
    "deterministic": True
}
  1. Add tests:
# tests/test_my_tool.py

import pytest
from edge_node.tools.my_tool import execute, MyToolInput

@pytest.mark.asyncio
async def test_my_tool():
    input = MyToolInput(input_field="hello")
    result = await execute(input)
    assert result.output_field == "HELLO"
  1. Document the tool:

Update docs/api/tool-catalog.md with tool details.

Adding a New Policy

  1. Create policy file:
# runtime/policies/my_policy.py

from runtime.policies.base import Policy, PolicyResult
from runtime.workflow import Workflow

class MyPolicy(Policy):
    """Description of what the policy enforces."""

    def check_workflow(self, workflow: Workflow) -> PolicyResult:
        # Policy logic here
        if some_condition:
            return PolicyResult(
                ok=False,
                error="Policy violation message"
            )
        return PolicyResult(ok=True)
  1. Register policy:
# runtime/policies/__init__.py

from runtime.policies.my_policy import MyPolicy

POLICIES = {
    "my_policy": MyPolicy(),
    # ... other policies
}
  1. Add tests:
# tests/test_my_policy.py

from runtime.policies.my_policy import MyPolicy
from runtime.workflow import Workflow

def test_my_policy():
    policy = MyPolicy()
    workflow = Workflow(...)  # Create test workflow
    result = policy.check_workflow(workflow)
    assert result.ok

Testing Guidelines

Unit Tests

Test individual components in isolation:

# tests/test_router.py

import pytest
from gateway.router import Router

def test_router_selects_best_node():
    router = Router()
    nodes = [...]  # Create test nodes
    selected = router.select_node(nodes, "pii_redact")
    assert selected.node_id == "expected_node"

Integration Tests

Test components working together:

# tests/integration/test_workflow_execution.py

import pytest
from gateway.http_gateway import app
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_workflow_execution():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post("/v1/tool/call", json={
            "target": {"kind": "local"},
            "tool_name": "pii_redact",
            "tool_args": {"text": "test@example.com"}
        })
        assert response.status_code == 200
        assert response.json()["ok"]

Test Fixtures

Use pytest fixtures for common setup:

# tests/conftest.py

import pytest
from gateway.router import Router

@pytest.fixture
def router():
    return Router()

@pytest.fixture
def mock_nodes():
    return [
        Node(node_id="node_a", tools=["pii_redact"]),
        Node(node_id="node_b", tools=["summarize"])
    ]

Documentation

Updating Documentation

  • Documentation is in docs/ directory
  • Written in Markdown
  • Built with MkDocs
# Install MkDocs
pip install mkdocs mkdocs-material

# Serve documentation locally
mkdocs serve

# Build documentation
mkdocs build

Documentation Structure

docs/
├── index.md
├── getting-started/
│   ├── quickstart.md
│   └── installation.md
├── concepts/
│   ├── overview.md
│   └── ...
├── workflows/
│   ├── defining-workflows.md
│   └── ...
├── api/
│   ├── http-gateway.md
│   └── ...
└── deployment/
    ├── architecture.md
    └── ...

Pull Request Guidelines

Before Submitting

  • [ ] Tests pass (pytest)
  • [ ] Code is formatted (black . and isort .)
  • [ ] Type hints added (mypy . passes)
  • [ ] Documentation updated
  • [ ] Commit messages are descriptive

PR Template

## Description
Brief description of changes

## Type of Change
- [ ] Bug fix
- [ ] New feature
- [ ] Breaking change
- [ ] Documentation update

## Testing
Describe how you tested the changes

## Checklist
- [ ] Tests added/updated
- [ ] Documentation updated
- [ ] Code follows style guide
- [ ] All tests pass

Code Review Process

Review Criteria

  • Correctness: Does it work as intended?
  • Test Coverage: Are there adequate tests?
  • Code Quality: Is it readable and maintainable?
  • Documentation: Are changes documented?
  • Performance: Are there performance implications?

Review Timeline

  • Reviewers aim to respond within 48 hours
  • Address feedback and update PR
  • Once approved, maintainers will merge

Communication

Channels

  • GitHub Issues: Bug reports and feature requests
  • Pull Requests: Code contributions and discussions
  • Email: sales@adaptivesentience.io

Reporting Bugs

Use GitHub Issues with this template:

**Describe the bug**
Clear description of the issue

**To Reproduce**
Steps to reproduce:
1. ...
2. ...

**Expected behavior**
What you expected to happen

**Environment**
- OS: [e.g., macOS 12.0]
- Python version: [e.g., 3.11]
- Agent Mesh version: [e.g., 1.0.0]

**Additional context**
Any other relevant information

License

By contributing, you agree that your contributions will be licensed under the project's proprietary license.

Next Steps