agent-factory-core

Agent Factory Core

Core library for building AI agents with pluggable models and channels.

Installation

# Basic installation
pip install agent-factory-core

# With Bedrock support
pip install agent-factory-core[bedrock]

# With Anthropic API support
pip install agent-factory-core[anthropic]

# With Slack channel support
pip install agent-factory-core[slack]

# With Teams channel support
pip install agent-factory-core[teams]

# All providers
pip install agent-factory-core[all]

Quick Start

Basic Agent

from agent_factory import AgentOrchestrator
from agent_factory.models import BedrockModel
from agent_factory.tools import tool

# Define a tool
@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Weather in {city}: 72°F, Sunny"

# Create orchestrator with Bedrock
orchestrator = AgentOrchestrator(
    model=BedrockModel(model_id="anthropic.claude-3-5-sonnet-20241022-v2:0"),
    system_prompt="You are a helpful weather assistant.",
    tools=[get_weather],
)

# Invoke
response = orchestrator.invoke(
    prompt="What's the weather in Miami?",
    session_id="user-123",
)
print(response["content"])

Using Anthropic API

from agent_factory import AgentOrchestrator
from agent_factory.models import AnthropicModel

orchestrator = AgentOrchestrator(
    model=AnthropicModel(
        model_id="claude-3-5-sonnet-20241022",
        api_key="sk-ant-...",  # Or use ANTHROPIC_API_KEY env var
    ),
    system_prompt="You are a helpful assistant.",
)

Slack Channel Integration

from agent_factory.channels import SlackChannel

# Create Slack channel
slack = SlackChannel(
    bot_token="xoxb-...",
    signing_secret="...",
)

# Handle incoming event
response = slack.handle_event(event_body)

# Send message
slack.send_message(
    channel="#general",
    text="Hello from the agent!",
)

Teams Channel Integration

from agent_factory.channels import TeamsChannel

# Create Teams channel
teams = TeamsChannel(
    app_id="your-app-id",        # Or use TEAMS_APP_ID env var
    app_password="your-secret",   # Or use TEAMS_APP_PASSWORD env var
)

# Handle incoming activity
message = teams.handle_event(activity)

# Send response
teams.send_message(
    conversation_id=message.channel_id,
    text="Hello from the bot!",
    service_url=message.metadata["service_url"],
)

# Send Adaptive Card
teams.send_adaptive_card(
    conversation_id=message.channel_id,
    card={"type": "AdaptiveCard", "body": [...]},
    service_url=message.metadata["service_url"],
)

Lambda Handler

from agent_factory import AgentOrchestrator
from agent_factory.models import BedrockModel
from agent_factory.handlers import LambdaHandler

orchestrator = AgentOrchestrator(
    model=BedrockModel(),
    system_prompt="You are a helpful assistant.",
)
handler = LambdaHandler(orchestrator)

def lambda_handler(event, context):
    return handler.handle(event)

Architecture

Models

The library provides a pluggable model interface:

• BedrockModel - AWS Bedrock (Claude, Llama, etc.)
• AnthropicModel - Direct Anthropic API

Create custom models by extending BaseModel:

from agent_factory.models import BaseModel, ModelResponse

class CustomModel(BaseModel):
    def converse(self, messages, system_prompt, tools=None, **kwargs):
        # Your implementation
        return ModelResponse(content="...", stop_reason=StopReason.END_TURN)

Channels

Supported channels:

• SlackChannel - Slack integration
• TeamsChannel - Microsoft Teams integration
• API - Use LambdaHandler with API Gateway

Channel Allowlist

Restrict which Slack channels can interact with your agent:

from agent_factory.channels import SlackChannel

# Only allow specific channels
slack = SlackChannel(
    allowed_channels=["C0123456789", "C9876543210"],
)

# Or use environment variable (comma-separated)
# SLACK_ALLOWED_CHANNELS=C0123456789,C9876543210
slack = SlackChannel()  # Will read from env var

Messages from unauthorized channels will raise ChannelAccessDeniedError.

Tools

Define tools with the @tool decorator:

from agent_factory.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> list:
    """Search the database for records.

    Args:
        query: Search query string
        limit: Maximum results to return
    """
    return [{"id": 1, "name": "Result"}]

Configuration

License

MIT