AXMP AI Agent Core

A Python-based AI agent framework built on LangGraph and LangChain for creating intelligent, multi-LLM agent systems with support for Model Context Protocol (MCP) servers.

🚀 Features

• 🤖 Multi-LLM Support: Seamless integration with OpenAI, Anthropic, and Google AI models
• 🔗 LangGraph Orchestration: Advanced agent state management with persistent checkpointing
• 🛠️ MCP Server Integration: Native support for Model Context Protocol with stdio, SSE, and HTTP transports
• 💬 Conversation Management: Thread-based chat history with MongoDB persistence
• 🔄 Stateful Agents: PostgreSQL-backed checkpoint storage for conversation resumption
• 🎯 Node-Based Workflows: Visual workflow system with triggers (chatbot, webhook, scheduler)
• 🔐 RBAC & Authentication: Built-in role-based access control and user management
• 📦 Dependency Injection: Clean architecture with IoC container pattern
• ⚡ FastAPI Backend: High-performance REST API with SSE streaming support

📋 Table of Contents

• Installation
• Quick Start
• Architecture
• Configuration
• Development
• Testing
• API Documentation
• Contributing

📦 Installation

Prerequisites

• Python 3.12+
• MongoDB (for data persistence)
• PostgreSQL (for LangGraph checkpointing)
• Redis (optional, for caching)
• AWS S3 (optional, for file storage)

Using uv (Recommended)

# Clone the repository
git clone https://github.com/yourusername/axmp-ai-agent-core.git
cd axmp-ai-agent-core

# Install dependencies with uv
uv sync

# Install development dependencies
uv sync --group dev

# Activate virtual environment
source .venv/bin/activate

Using pip

pip install axmp-ai-agent-core

🎯 Quick Start

1. Environment Setup

Create a .env file in the project root:

# AI Service Keys
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...

# Core Settings
CORE_API_ENDPOINT=http://localhost:8000
CORE_WEB_ENDPOINT=http://localhost:3000
CORE_DEFAULT_MODEL=openai/gpt-4.1-mini

# MongoDB Configuration
MONGODB_HOSTNAME=localhost
MONGODB_PORT=27017
MONGODB_USERNAME=admin
MONGODB_PASSWORD=password
MONGODB_DATABASE=axmp_ai_agent

# PostgreSQL Configuration (for LangGraph checkpointing)
POSTGRESQL_HOSTNAME=localhost
POSTGRESQL_PORT=5432
POSTGRESQL_USERNAME=postgres
POSTGRESQL_PASSWORD=password
POSTGRESQL_DATABASE=langgraph_checkpoints

# AWS S3 (Optional)
# AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY are optional. Leave them unset to use
# the default AWS credential chain (EKS Pod Identity, IRSA, instance role, ...).
AWS_ACCESS_KEY_ID=your-access-key
AWS_SECRET_ACCESS_KEY=your-secret-key
AWS_DEFAULT_REGION=us-east-1
AWS_S3_BUCKET_NAME=axmp-ai-storage
AWS_S3_BUCKET_ROOT=agents

2. Basic Usage

from axmp_ai_agent_core.agent.wrapper import DeepAgentWrapper, DeepAgentContext

# Create DeepAgentWrapper with a DeepAgentSpecProfile
wrapper = DeepAgentWrapper(
    agent_spec=agent_spec_profile,
    oauth_mcp_server_registry=registry,
    mcp_client_factory=factory,
    checkpointer=checkpointer,
    store=store,
    context_schema=DeepAgentContext,
)

# Build and run the agent
graph = await wrapper.build_deepagent(user_id="user-123")

# Run conversation
response = await agent.agent.ainvoke(
    {"messages": [{"role": "user", "content": "Hello!"}]},
    config={"configurable": config.model_dump()}
)

3. Running the API Server

from fastapi import FastAPI
from axmp_ai_agent_core.router import agent_router, healthy_router
from axmp_ai_agent_core.di.service_container import ServicesContainer

app = FastAPI(title="AXMP AI Agent API")

# Initialize dependency injection
container = ServicesContainer()
container.wire(packages=["axmp_ai_agent_core.router"])

# Register routers
app.include_router(agent_router.router, prefix="/api/v1", tags=["agents"])
app.include_router(healthy_router.router, prefix="/api/v1", tags=["health"])

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

🏗️ Architecture

System Overview

┌─────────────────────────────────────────────────────────────┐
│                      FastAPI Routers                        │
│  (HTTP Layer - REST API Endpoints + SSE Streaming)         │
└─────────────────────┬───────────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────────┐
│                    Service Layer                            │
│  (Business Logic - AgentProfileService, UserService, etc.)  │
└─────────────────────┬───────────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────────┐
│                  Repository Layer                           │
│  (Data Access - MongoDB CRUD Operations)                    │
└─────────────────────┬───────────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────────┐
│                    Entity Models                            │
│  (Domain Models - Pydantic Schemas)                         │
└─────────────────────────────────────────────────────────────┘

Core Components

Agent System

• DeepAgentWrapper: Multi-agent orchestration wrapper using deepagents library
• DeepAgentContext: Context dataclass for DeepAgent execution

Dependency Injection

The project uses a three-layer DI container architecture:

1. ResourcesContainer: External resources (MongoDB, PostgreSQL, Redis, S3)
2. RepositoriesContainer: Data access layer repositories
3. ServicesContainer: Business logic services

Entity Models

• AgentProfile: Node-based agent workflow definitions
• ChatConversation: Thread-based conversation history
• ChatMemory: Persistent agent memory
• LlmProvider: LLM provider credentials
• UserCredential: User-specific service credentials

Agent Workflow System

Agents use a visual node-based workflow system:

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│   Trigger    │────▶│   LLM Node   │────▶│   Tool Node  │
│   (Entry)    │     │ (Processing) │     │  (Actions)   │
└──────────────┘     └──────────────┘     └──────────────┘
                             │
                             ▼
                     ┌──────────────┐
                     │   End Node   │
                     │   (Output)   │
                     └──────────────┘

Node Types:

• Trigger Nodes: Chatbot, Webhook, Scheduler (with cron expressions)
• LLM Nodes: Execute language model inference with tool/MCP bindings
• Tool Nodes: External API and service integrations
• Subflow Nodes: Nested workflow execution
• End Nodes: Terminal output nodes

⚙️ Configuration

Core Settings

All configuration is managed through Pydantic settings classes in setting.py:

AI Service Configuration

from axmp_ai_agent_core.setting import ai_service_key_settings

# Access API keys
openai_key = ai_service_key_settings.openai_api_key
anthropic_key = ai_service_key_settings.anthropic_api_key

Core Application Settings

from axmp_ai_agent_core.setting import core_settings

# Agent configuration
default_model = core_settings.default_model  # "openai/gpt-4.1-mini"
max_tokens = core_settings.default_model_max_tokens  # 5000
recursion_limit = core_settings.recursion_limit  # 30

# Caching configuration
agent_cache_ttl = core_settings.agent_cache_ttl  # 3600 seconds
agent_cache_max_size = core_settings.agent_cache_max_size  # 100 agents

Database Configuration

from axmp_ai_agent_core.setting import mongodb_settings, postgresql_settings

# MongoDB connection
mongo_uri = mongodb_settings.uri

# PostgreSQL connection (for LangGraph)
postgres_uri = postgresql_settings.db_uri_with_params

🛠️ Development

Code Quality

# Run linter with auto-fix
ruff check . --fix

# Run formatter
ruff format .

# Run pre-commit hooks
pre-commit run --all-files

# Install pre-commit hooks
pre-commit install

Code Style Guidelines

• Docstrings: Google-style docstrings required (enforced by ruff)
• First Line: Must be in imperative mood (e.g., "Create user" not "Creates user")
• Type Hints: Modern Python 3.12+ syntax required
• Import Sorting: Automatic via ruff (isort-compatible)

🧪 Testing

Running Tests

# Run all tests
uv run pytest

# Run with verbose output
uv run pytest -v

# Run with coverage report
uv run pytest --cov=src/axmp_ai_agent_core --cov-report=html

# Run specific directory
uv run pytest tests/entity/ -v        # Entity model tests only
uv run pytest tests/k8s/ -v           # K8s tests only

# Run specific test file
uv run pytest tests/k8s/test_k8s_utils.py -v

# Run specific test class or function
uv run pytest tests/k8s/test_k8s_utils.py::TestSanitizeKubernetesName -v
uv run pytest tests/k8s/test_k8s_utils.py::TestSanitizeKubernetesName::test_valid_name_unchanged -v

# Fail fast (stop on first failure)
uv run pytest -x

# Watch mode (auto-rerun on file changes)
uv run pytest-watcher

Test Structure

tests/
├── conftest.py                          # Shared fixtures (sample dicts, model instances)
├── entity/                              # Entity model tests (185 tests)
│   ├── test_agent_profile.py            #   AgentProfile model ↔ dict
│   ├── test_base_model.py               #   CoreBaseModel, NamedCoreBaseModel
│   ├── test_kubernetes_config.py        #   KubernetesConfig model
│   ├── test_mcp_server_profile.py       #   McpServerProfile model
│   ├── test_shared_target.py            #   SharedTarget, RBAC models
│   └── test_user_rbac.py               #   User, Role, Permission models
└── k8s/                                 # Kubernetes integration tests (187 tests)
    ├── conftest.py                      #   K8s mock fixtures (manager factory, API mocks)
    ├── model/                           #   K8s model tests (105 tests)
    │   ├── test_custom_resource.py      #     CRD, Phase, Endpoints, Metadata
    │   ├── test_base_instance.py        #     BaseInstance, Spec, Status
    │   ├── test_agent_instance.py       #     AgentInstance model
    │   └── test_mcp_server_instance.py  #     McpServerInstance model
    ├── test_k8s_utils.py                #   Utility functions (37 tests)
    └── test_k8s_resource_manager.py     #   K8sResourceManager async CRUD (45 tests)

Test Configuration

The project uses:

• pytest with pytest-asyncio (strict mode) for async test support
• pytest-cov for coverage tracking
• pytest-watcher for watch mode
• unittest.mock for K8s API mocking
• Warnings filtered for cleaner output

Test Conventions

All tests follow the AAA (Arrange-Act-Assert) pattern with explicit comments:

def test_example(self):
    """Descriptive test name."""
    # Arrange
    data = {"name": "test-agent", "namespace": "default"}

    # Act
    result = AgentInstance.model_validate(data)

    # Assert
    assert result.metadata.name == "test-agent"

For async tests (e.g., K8sResourceManager), use @pytest.mark.asyncio:

@pytest.mark.asyncio
async def test_create_success(self, agent_manager, mock_custom_api, sample_agent_dict):
    """K8s resource creation test."""
    # Arrange
    mock_custom_api.create_namespaced_custom_object.return_value = sample_agent_dict

    # Act
    result = await agent_manager.create(instance=instance, namespace="default")

    # Assert
    assert result.metadata.name == "test-agent"

📚 API Documentation

Key Endpoints

Agent Profile Management

GET    /api/v1/ai-agents              # List agent profiles
POST   /api/v1/ai-agents              # Create agent profile
GET    /api/v1/ai-agents/{id}         # Get agent profile
PUT    /api/v1/ai-agents/{id}         # Update agent profile
DELETE /api/v1/ai-agents/{id}         # Delete agent profile

Chat Conversations

POST   /api/v1/ai-agents/{agent_id}/conversations/{thread_id}/chat
       # Send message to agent (SSE streaming response)

POST   /api/v1/ai-agents/{agent_id}/conversations/{thread_id}/title
       # Generate conversation title

GET    /api/v1/ai-agents/{agent_id}/conversations/{thread_id}
       # Get conversation history

Health Check

GET    /api/v1/health                 # Health check endpoint

SSE Streaming Response

Chat endpoints return Server-Sent Events for real-time streaming:

const eventSource = new EventSource(
  '/api/v1/ai-agents/agent-123/conversations/thread-456/chat'
);

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data.content);
};

🤝 Contributing

Development Workflow

1. Fork the repository
2. Create a feature branch: git checkout -b feature/amazing-feature
3. Make your changes
4. Run tests: pytest
5. Run linter: ruff check . --fix
6. Commit your changes: git commit -m 'Add amazing feature'
7. Push to the branch: git push origin feature/amazing-feature
8. Open a Pull Request

Commit Message Convention

Follow conventional commits:

feat: add MCP server support
fix: resolve conversation persistence issue
docs: update README with examples
test: add agent initialization tests
refactor: improve repository pattern

Pre-commit Hooks

The project uses pre-commit hooks to ensure code quality:

• Trailing whitespace removal
• End-of-file fixer
• Debug statement detection
• Ruff linting and formatting

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

👥 Authors

• Kilsoo Kang - Initial work - kilsoo75@gmail.com

🙏 Acknowledgments

• Built with LangChain and LangGraph
• Powered by FastAPI
• Code quality ensured by Ruff

📞 Support

For questions and support:

• 📧 Email: kilsoo75@gmail.com
• 🐛 Issues: GitHub Issues

Made with ❤️ by the AXMP Team

Error codes

Every backend built on this package answers errors with one envelope ({"result":"failed","code":"E201","message":"...","data":...,"error_id":...}). Register the handlers once per app and raise CoreServiceException with a CoreError (or your app's own Error enum):

from axmp_ai_agent_core.exception.handlers import install_error_handlers
install_error_handlers(app)

The code table is exported from the enums, never written by hand:

axmp-error-codes --format markdown                      # docs table (E + G)
axmp-error-codes --enum my_app.exception.errors:AppError --format i18n -o error-codes.json

See docs/error-codes.md for the scheme, the bands and the full table.