arp/arp_agent

ARP Agent - LLM Agent for the ARP Platform

An LLM-powered agent that connects to the ARP (Agent-native ERP) platform via the Model Context Protocol (MCP) and responds to Task and Message events in real-time.

Table of Contents

• Overview
• Installation
• Configuration
• Running the Agent
• How It Works
• MCP Communication
• Available MCP Tools
• Workflows
• Programmatic Usage
• Testing

---

Overview

The ARP Agent connects to an ARP server and:

1. Authenticates via GraphQL login to obtain a JWT token
2. Connects to MCP via Server-Sent Events (SSE) to the /mcp endpoint
3. Discovers Tools using the MCP tools/list protocol
4. Subscribes to Events via MCP resources for real-time notifications
5. Processes Events using an LLM with tool-calling capabilities

---

Installation

Using pip

pip install -r requirements.txt

Using Poetry (recommended)

poetry install

Dependencies

• openai - OpenAI API client for LLM interactions
• requests - HTTP client for GraphQL and MCP communication
• sseclient-py - Server-Sent Events client for MCP
• python-dotenv - Environment variable management

---

Configuration

Copy the example environment file and configure your credentials:

cp .env.example .env

Edit .env with your settings:

# ARP Server Configuration
ARP_URL=http://localhost:8080
ARP_USERNAME=your-email@example.com
ARP_PASSWORD=your-password

# OpenAI Configuration
OPENAI_API_KEY=sk-your-openai-api-key
OPENAI_MODEL=gpt-4
OPENAI_TEMPERATURE=0.0

# Optional: Custom OpenAI endpoint (for local models, etc.)
# OPENAI_BASE_URL=http://localhost:11434/v1

Required Environment Variables

Variable	Description
ARP_URL	Base URL of the ARP server
ARP_USERNAME	Your ARP login email
ARP_PASSWORD	Your ARP password
OPENAI_API_KEY	OpenAI API key for LLM

Optional Environment Variables

Variable	Description	Default
OPENAI_MODEL	Model to use	gpt-4
OPENAI_TEMPERATURE	Sampling temperature	0.0
OPENAI_BASE_URL	Custom OpenAI-compatible endpoint	OpenAI API

---

Running the Agent

python run_arp_agent.py

Expected Output

Testing connectivity to OpenAI API (api.openai.com)...
✓ Successfully connected to OpenAI API (api.openai.com)
Connecting to ARP server at http://localhost:8080...
Successfully authenticated with ARP server
Connecting to MCP server...
Discovered 3 MCP tools: ['introspect', 'query', 'mutate']
Initializing LLM agent...
Agent initialized successfully.

Subscribing to ARP resources...
Available resources: ['taskCreated', 'taskUpdated', 'taskDeleted', 'messageAdded']
  Subscribed to: graphql://subscription/taskCreated
  Subscribed to: graphql://subscription/taskUpdated
  Subscribed to: graphql://subscription/taskDeleted
  Subscribed to: graphql://subscription/messageAdded

Listening for events. Press Ctrl+C to stop.

---

How It Works

Architecture

┌─────────────────┐      GraphQL Login      ┌─────────────────┐
│   ARP Agent     │ ──────────────────────► │   ARP Server    │
│                 │ ◄────────────────────── │                 │
│  ┌───────────┐  │      JWT Token          │                 │
│  │    LLM    │  │                         │  ┌───────────┐  │
│  └───────────┘  │      SSE Connect        │  │    MCP    │  │
│        │        │ ──────────────────────► │  │  Server   │  │
│        ▼        │      /mcp endpoint      │  └───────────┘  │
│  ┌───────────┐  │                         │        │        │
│  │MCP Client │◄─┼──── Tool Discovery      │        │        │
│  └───────────┘  │      Tool Calls         │        ▼        │
│        │        │      Notifications      │  ┌───────────┐  │
│        ▼        │                         │  │ GraphQL   │  │
│  ┌───────────┐  │                         │  │  Engine   │  │
│  │  Events   │◄─┼─────────────────────────┼──┤           │  │
│  └───────────┘  │      Real-time          │  └───────────┘  │
└─────────────────┘      Subscriptions      └─────────────────┘

Event Processing Flow

1. Event Received: Task or message event via MCP resource notification
2. Context Built: Extract relevant details (ID, title, content, sender, etc.)
3. LLM Invoked: Agent processes the event with available tools
4. Tool Execution: LLM may call MCP tools to query/mutate data
5. Response Generated: Agent produces a result or takes action

---

MCP Communication

The agent uses the Model Context Protocol (MCP) to communicate with the ARP server:

Connection Flow

1. SSE Connection: Connect to /mcp endpoint via Server-Sent Events
2. Endpoint Discovery: Receive message endpoint URL from endpoint event
3. Initialize: Send initialize request with protocol version and client info
4. Tool Discovery: Call tools/list to discover available tools
5. Subscribe: Call resources/subscribe for real-time event streams

Authentication

Authentication is handled via JWT tokens:

1. Login via GraphQL login mutation with email/password
2. Receive JWT token in response
3. Include token in SSE connection headers (Authorization: Bearer <token>)
4. Token is automatically propagated to all MCP requests

Protocol Details

• Protocol Version: 2024-11-05
• Transport: HTTP POST for requests, SSE for responses/notifications
• Format: JSON-RPC 2.0

---

Available MCP Tools

The ARP MCP server exposes three tools:

1. introspect

Discover the GraphQL schema - types, fields, queries, mutations.

# Get full schema
result = mcp_client.call_tool("introspect", {})

# Get specific type
result = mcp_client.call_tool("introspect", {"typeName": "User"})

2. query

Execute GraphQL queries (read operations).

# Query users
result = mcp_client.call_tool("query", {
    "query": "{ users { id email roles { name } } }"
})

# Query with variables
result = mcp_client.call_tool("query", {
    "query": "query User($id: ID!) { user(id: $id) { id email } }",
    "variables": {"id": "1"}
})

3. mutate

Execute GraphQL mutations (create/update/delete operations).

# Create a task
result = mcp_client.call_tool("mutate", {
    "mutation": """
        mutation CreateTask($input: NewTask!) {
            createTask(input: $input) { id title }
        }
    """,
    "variables": {
        "input": {
            "title": "New Task",
            "content": "Task description",
            "createdById": "1"
        }
    }
})

# Delete a note
result = mcp_client.call_tool("mutate", {
    "mutation": "mutation DeleteNote($id: ID!) { deleteNote(id: $id) }",
    "variables": {"id": "123"}
})

---

Workflows

The ARP platform supports workflows - configurable, DAG-based process automation that coordinates tasks across agents and users. Workflows enable you to define multi-step processes with dependencies, parallel execution, and automatic task creation.

Workflow Concepts

Concept	Description
WorkflowTemplate	Admin-defined workflow definition (JSON DAG structure)
WorkflowInstance	A running instance of a workflow template
WorkflowNode	A single step in a workflow instance (maps to a Task)
WorkflowEdge	Dependency relationship between nodes

Node Types

Type	Description
task	Creates and assigns a task to a user
condition	Conditional branching based on workflow context
parallel	Fork into multiple concurrent branches
join	Wait for multiple branches to complete
trigger	External event trigger (webhook, schedule, etc.)

Node Status Lifecycle

pending → ready → running → completed
                  └── failed → (retry or abort)
                  └── skipped

• pending: Waiting for dependencies to complete
• ready: All dependencies satisfied, ready to execute
• running: Currently executing (task created)
• completed: Successfully finished
• failed: Execution failed (may retry)
• skipped: Conditionally bypassed

Workflow Definition Format

Workflows are defined as JSON DAGs (Directed Acyclic Graphs):

{
  "nodes": {
    "start": {
      "type": "task",
      "title": "Initial Review",
      "content": "Review the submitted request",
      "assignee": "reviewer@example.com",
      "dependsOn": []
    },
    "parallel_analysis": {
      "type": "parallel",
      "title": "Parallel Analysis",
      "content": "Run multiple analyses in parallel",
      "dependsOn": ["start"]
    },
    "technical_review": {
      "type": "task",
      "title": "Technical Review",
      "content": "Review technical aspects",
      "assignee": "tech@example.com",
      "dependsOn": ["parallel_analysis"]
    },
    "business_review": {
      "type": "task",
      "title": "Business Review",
      "content": "Review business impact",
      "assignee": "business@example.com",
      "dependsOn": ["parallel_analysis"]
    },
    "join_reviews": {
      "type": "join",
      "title": "Join Reviews",
      "content": "Wait for all reviews to complete",
      "dependsOn": ["technical_review", "business_review"]
    },
    "final_approval": {
      "type": "task",
      "title": "Final Approval",
      "content": "Make final decision",
      "assignee": "approver@example.com",
      "dependsOn": ["join_reviews"]
    }
  }
}

Creating Workflows via MCP

Use the mutate tool to create workflow templates and start instances:

# Create a workflow template
result = mcp_client.call_tool("mutate", {
    "mutation": """
        mutation CreateWorkflowTemplate($input: NewWorkflowTemplate!) {
            createWorkflowTemplate(input: $input) {
                id
                name
                isActive
            }
        }
    """,
    "variables": {
        "input": {
            "name": "Approval Process",
            "description": "Multi-step approval workflow",
            "definition": '{"nodes": {...}}',
            "isActive": True
        }
    }
})

# Start a workflow instance
result = mcp_client.call_tool("mutate", {
    "mutation": """
        mutation StartWorkflow($templateId: ID!, $input: StartWorkflowInput!) {
            startWorkflow(templateId: $templateId, input: $input) {
                id
                status
                createdAt
            }
        }
    """,
    "variables": {
        "templateId": "1",
        "input": {
            "serviceId": "5",
            "context": '{"requestId": "REQ-123"}'
        }
    }
})

Querying Workflow State

# List all workflow templates
result = mcp_client.call_tool("query", {
    "query": """
        {
            workflowTemplates {
                id
                name
                description
                isActive
            }
        }
    """
})

# Get workflow instance status
result = mcp_client.call_tool("query", {
    "query": """
        query WorkflowInstance($id: ID!) {
            workflowInstance(id: $id) {
                id
                status
                context
                service { id name }
                template { name }
            }
        }
    """,
    "variables": {"id": "1"}
})

Workflow Execution Flow

1. Template Created: Admin defines workflow structure
2. Instance Started: Workflow instance created from template
3. Root Nodes Execute: Nodes with no dependencies create tasks
4. Dependencies Resolve: As tasks complete, downstream nodes become ready
5. Parallel Branches: Multiple branches execute concurrently
6. Join Points: Wait for all incoming branches to complete
7. Completion: Workflow marked complete when all nodes finish

Agent Interaction with Workflows

Agents can interact with workflows through MCP tools:

• Query workflow templates and instances to understand current state
• Create workflow templates for new processes
• Start workflow instances when triggered by events
• Update task status to progress workflow nodes
• Monitor workflow completion and handle failures

Example agent workflow handling:

# Agent receives task completion event
# Check if task is part of a workflow
result = mcp_client.call_tool("query", {
    "query": """
        query TaskWorkflow($taskId: ID!) {
            task(id: $taskId) {
                id
                title
                workflowNodes {
                    id
                    workflowInstance {
                        id
                        status
                        template { name }
                    }
                }
            }
        }
    """,
    "variables": {"taskId": task_id}
})

# If task is part of workflow, check downstream nodes
# Agent can proactively notify next assignees or take actions

---

Programmatic Usage

Using MCPClient Directly

from llm_agents.mcp_client import login_and_create_mcp_client

# Login and create client
client = login_and_create_mcp_client(
    url="http://localhost:8080",
    username="admin@example.com",
    password="secret123"
)

# Connect to MCP server
client.connect()
client.initialize()

# Discover tools
tools = client.list_tools()
print(f"Available tools: {[t.name for t in tools]}")

# Call tools
users = client.call_tool("query", {"query": "{ users { id email } }"})
print(users)

# Subscribe to resources
client.subscribe_resource("graphql://subscription/taskCreated")

# Listen for notifications
def on_notification(data):
    print(f"Received: {data}")

client.listen_for_notifications(on_notification)

# Cleanup
client.close()

Using the Agent with MCP

from llm_agents import Agent, ChatLLM
from llm_agents.mcp_client import login_and_create_mcp_client

# Create authenticated MCP client
mcp_client = login_and_create_mcp_client(
    url="http://localhost:8080",
    username="admin@example.com",
    password="secret123"
)

# Connect and initialize
mcp_client.connect()
mcp_client.initialize()
mcp_client.list_tools()

# Create agent with MCP client
llm = ChatLLM()  # Uses OPENAI_API_KEY from environment
agent = Agent(llm=llm, mcp_client=mcp_client)

# Run the agent
result = agent.run("List all users and their roles")
print(result)

# Cleanup
mcp_client.close()

---

MCP Resources (Subscriptions)

The ARP MCP server exposes resources for real-time GraphQL subscriptions:

Resource URI	Description
graphql://subscription/taskCreated	New task events (received by assignee)
graphql://subscription/taskUpdated	Task update events (received by assignee)
graphql://subscription/taskDeleted	Task deletion events (received by assignee)
graphql://subscription/messageAdded	New message events (received by receivers)

Subscribing to Resources

# List available resources
resources = mcp_client.list_resources()
for resource in resources:
    print(f"{resource['uri']}: {resource['name']}")

# Subscribe to task events
mcp_client.subscribe_resource("graphql://subscription/taskCreated")
mcp_client.subscribe_resource("graphql://subscription/taskUpdated")

# Unsubscribe
mcp_client.unsubscribe_resource("graphql://subscription/taskCreated")

---

Event Filtering

Events are filtered by the ARP server based on user context:

• Task Events: Only received for tasks where the user is the assignee
• Message Events: Only received for messages where the user is a receiver

This ensures each user only receives relevant notifications.

---

Testing

Run the integration tests (requires a running ARP server):

python -m pytest tests/integration/ -v --no-cov

Test Categories

• Login Tests: Authentication flow
• Connection Tests: MCP connection and initialization
• Tool Tests: Introspect, query, and mutate operations
• Resource Tests: Subscription functionality
• Error Handling Tests: Error responses and edge cases

---

Project Structure

arp_agent/
├── llm_agents/
│   ├── __init__.py          # Package exports
│   ├── agent.py             # Agent with tool-calling
│   ├── llm.py               # OpenAI LLM wrapper
│   └── mcp_client.py        # MCP client implementation
├── tests/
│   ├── conftest.py          # Pytest fixtures
│   ├── test_setup_validation.py
│   ├── unit/                # Unit tests
│   └── integration/         # Integration tests
│       └── test_arp_agent_integration.py
├── specs/
│   └── schema.graphqls      # GraphQL schema reference
├── run_arp_agent.py         # Main entry point
├── run_tests.py             # Test runner wrapper
├── pyproject.toml           # Poetry configuration
├── poetry.lock              # Locked dependencies
├── requirements.txt         # Python dependencies (for pip)
├── .env.example             # Environment template
├── CLIENT_GUIDE.md          # ARP client implementation guide
└── README.md                # This file

---

License

See LICENSE for details.