Core Concepts

​

Overview

​

Entrypoints

​

What are Entrypoints?

def my_agent(message: str, user_id: str = "default") -> str:
    """This is an entrypoint function"""
    return f"Hello {user_id}, you said: {message}"

​

Entrypoint Characteristics

• Function Signature: Defines the API contract for all languages
• Parameters: Automatically become API parameters
• Return Value: Becomes the response sent to clients
• Documentation: Function docstrings become API documentation

​

Entrypoint Types

​

Synchronous Entrypoints

def sync_agent(query: str) -> Dict[str, Any]:
    """Synchronous agent that returns complete response"""
    return {
        "response": f"Processed: {query}",
        "status": "success",
        "timestamp": datetime.now().isoformat()
    }

​

Streaming Entrypoints

def stream_agent(query: str) -> Iterator[str]:
    """Streaming agent that yields response chunks"""
    yield f"Starting to process: {query}\n"
    yield "Analyzing your request...\n"
    yield "Generating response...\n"
    yield f"Final result: {query.upper()}\n"

​

Entrypoint Naming Convention

• Synchronous: Use descriptive names like chat, analyze, process
• Streaming: End with _stream suffix like chat_stream, analyze_stream

​

Tags

​

What are Tags?

{
  "entrypoints": [
    {
      "file": "main.py",
      "module": "chat_agent",
      "tag": "chat"
    },
    {
      "file": "main.py", 
      "module": "chat_agent_stream",
      "tag": "chat_stream"
    }
  ]
}

​

Tag Best Practices

• Use descriptive, lowercase names
• Avoid special characters and spaces
• Use underscores for multi-word tags
• Keep tags consistent across environments

​

Agent Lifecycle

​

1. Initialization

1. Configuration Loading: Reads runagent.config.json
2. Entrypoint Discovery: Imports and validates entrypoint functions
3. Server Startup: Starts REST and WebSocket servers
4. Health Check: Verifies all entrypoints are callable

​

2. Request Processing

1. Request Validation: Validates parameters against function signature
2. Function Invocation: Calls the appropriate entrypoint function
3. Response Processing: Handles return value or streaming
4. Client Response: Sends response back to client

​

3. Shutdown

1. Graceful Shutdown: Finishes processing current requests
2. Resource Cleanup: Closes connections and frees resources
3. Server Stop: Shuts down REST and WebSocket servers

​

Multi-Language Translation

​

How Function Signatures Become APIs

# Python function
def process_data(data: List[Dict], analysis_type: str = "summary") -> Dict[str, Any]:
    pass

# Python SDK
result = client.run(data=[{"key": "value"}], analysis_type="summary")

// JavaScript SDK
const result = await client.run({
    data: [{"key": "value"}],
    analysisType: "summary"
});

// Go SDK
result, err := client.Run(ctx, map[string]interface{}{
    "data": []map[string]interface{}{{"key": "value"}},
    "analysisType": "summary",
})

// Rust SDK
let result = client.run(&[
    ("data", json!([{"key": "value"}])),
    ("analysis_type", json!("summary")),
]).await?;

​

Type Translation Rules

​

Streaming Architecture

​

How Streaming Works

1. Client Request: Client calls streaming entrypoint
2. WebSocket Connection: Establishes real-time connection
3. Iterator Processing: Python generator yields chunks
4. Real-time Transmission: Chunks sent immediately to client
5. Native Iteration: Client receives chunks in language-appropriate way

​

Streaming Patterns

​

Python Generator

def stream_agent(query: str) -> Iterator[str]:
    yield f"Processing: {query}\n"
    yield "Step 1: Analyzing...\n"
    yield "Step 2: Generating...\n"
    yield f"Result: {query.upper()}\n"

​

JavaScript Async Iterator

const stream = await client.run({query: "Hello"});
for await (const chunk of stream) {
    console.log(chunk);
}

​

Rust Futures Stream

let mut stream = client.run_stream(&[("query", json!("Hello"))]).await?;
while let Some(chunk) = stream.next().await {
    print!("{}", chunk?);
}

​

Configuration System

​

runagent.config.json Structure

{
  "agent_name": "my-agent",
  "description": "A description of what this agent does",
  "framework": "custom",
  "agent_architecture": {
    "entrypoints": [
      {
        "file": "main.py",
        "module": "function_name",
        "tag": "endpoint_name"
      }
    ]
  }
}

​

Configuration Validation

1. File Existence: Checking that specified files exist
2. Module Import: Verifying modules can be imported
3. Function Existence: Confirming functions exist in modules
4. Signature Validation: Ensuring functions are callable
5. Tag Uniqueness: Verifying tags are unique

​

Error Handling

​

Error Types

​

Client-Side Errors

• AuthenticationError: Invalid or missing API key
• AgentNotFoundError: Agent doesn’t exist
• ValidationError: Invalid request parameters
• RateLimitError: Too many requests
• TimeoutError: Request timed out
• NetworkError: Connection issues

​

Server-Side Errors

• ImportError: Cannot import entrypoint module
• FunctionError: Entrypoint function raised exception
• ConfigurationError: Invalid configuration
• ServerError: Internal server error

​

Error Propagation

def my_agent(query: str) -> str:
    if not query:
        raise ValueError("Query cannot be empty")
    return f"Processed: {query}"

# Python SDK
try:
    result = client.run(query="")
except ValidationError as e:
    print(f"Validation error: {e}")

​

Security Model

​

Sandboxing

1. Process Isolation: Each agent runs in its own process
2. Resource Limits: CPU and memory constraints
3. Network Isolation: Controlled network access
4. File System: Restricted file system access

​

API Security

• Authentication: API key-based authentication
• Authorization: Role-based access control
• Rate Limiting: Request rate limiting
• Input Validation: Parameter validation and sanitization

​

Performance Considerations

​

Scalability

1. Horizontal Scaling: Multiple agent instances
2. Load Balancing: Automatic request distribution
3. Auto-scaling: Dynamic instance management
4. Caching: Response caching for performance

​

Optimization Tips

1. Efficient Entrypoints: Keep functions focused and fast
2. Streaming: Use streaming for long-running operations
3. Caching: Cache expensive computations
4. Resource Management: Monitor memory and CPU usage

​

Common Patterns

​

1. Simple Chat Agent

def chat_agent(message: str, user_id: str = "default") -> str:
    """Simple chat agent"""
    return f"Hello {user_id}, you said: {message}"

​

2. Data Processing Agent

def process_data(data: List[Dict], operation: str = "sum") -> Dict[str, Any]:
    """Process data with different operations"""
    if operation == "sum":
        return {"result": sum(item.get("value", 0) for item in data)}
    elif operation == "count":
        return {"result": len(data)}
    else:
        return {"error": "Unknown operation"}

​

3. Streaming Analysis Agent

def analyze_stream(data: str) -> Iterator[str]:
    """Streaming analysis agent"""
    yield "Starting analysis...\n"
    yield f"Processing {len(data)} characters\n"
    yield "Analyzing content...\n"
    yield f"Analysis complete: {data.upper()}\n"

​

Troubleshooting

​

Common Issues

1. Import Errors: Check Python path and dependencies
2. Function Not Found: Verify module and function names
3. Type Errors: Ensure parameter types match
4. Streaming Issues: Check iterator return type
5. Configuration Errors: Validate JSON syntax

​

Debug Tips

1. Enable Logging: Use verbose logging for debugging
2. Test Locally: Test functions before deploying
3. Check Dependencies: Ensure all imports are available
4. Validate Configuration: Use configuration validation tools
5. Monitor Performance: Track response times and resource usage

​

Next Steps