# Arc Runtime
[](https://www.python.org/downloads/)
[](https://opensource.org/licenses/MIT)
[](https://badge.fury.io/py/arc-runtime)
[](docs/performance_report.md)
[](https://opentelemetry.io/)
Arc Runtime is a lightweight Python interceptor that prevents AI agent failures in real-time by applying learned fixes before requests reach the LLM. It's the client-side component of the Arc AI reliability system, designed to work with Arc Core.
## Key Features
- **Zero-config interception** - Just import and it works
- **Ultra-low latency** - 0.011ms P99 overhead (99.78% better than 5ms requirement)
- **Thread-safe** - Works seamlessly with async and multi-threaded applications
- **Pattern matching** - Real-time detection and fixing of known failure patterns
- **Multi-agent support** - Track complex agent pipelines with context handoffs
- **MCP interception** - Monitor Model Context Protocol communications
- **LangGraph integration** - Automatic tracking for LangGraph workflows
- **OpenTelemetry support** - Full agent telemetry capture (reasoning traces, tool calls, tokens)
- **Graceful degradation** - Never breaks your application if Arc Core is unreachable
- **Local metrics** - Prometheus endpoint at http://localhost:9090/metrics
## How It Works
Arc Runtime intercepts outgoing LLM API calls and:
1. Matches requests against known failure patterns (<1ms)
2. Applies fixes before the request reaches the LLM
3. Streams telemetry to Arc Core for continuous learning
4. Exposes metrics for monitoring
## System Architecture
Arc Runtime is the client-side component that sits in your application environment:
```mermaid
graph TB
subgraph "Your Application Environment"
App[Your AI Application]
Arc[Arc Runtime]
SDK[OpenAI/Anthropic SDK]
Cache[(Local Cache)]
App --> Arc
Arc --> Cache
Cache --> Arc
Arc --> SDK
end
SDK --> API[LLM API]
API --> SDK
SDK --> Arc
Arc --> App
subgraph "Arc Core Service"
Collector[gRPC Collector]
Detector[Failure Detector]
Registry[Pattern Registry]
Collector --> Detector
Detector --> Registry
end
Arc -.-> Collector
Registry -.-> Cache
style Arc fill:#4CAF50,stroke:#2E7D32,stroke-width:2px
style App fill:#2196F3,stroke:#1565C0,stroke-width:2px
style Cache fill:#FFB74D,stroke:#F57C00,stroke-width:2px
style Collector fill:#E1BEE7,stroke:#9C27B0,stroke-width:2px
style Registry fill:#FFCDD2,stroke:#D32F2F,stroke-width:2px
```
**Request Flow:**
1. Your AI Application makes an API call
2. Arc Runtime intercepts the request
3. Checks local cache for matching failure patterns
4. Applies fixes if patterns match
5. Forwards the (potentially modified) request to the LLM SDK
6. SDK sends request to LLM API
7. Response flows back through Arc Runtime to your application
8. Arc Runtime asynchronously streams telemetry to Arc Core
**Key Integration Points:**
- **Telemetry Streaming**: Arc Runtime streams all request/response data to Arc Core via gRPC
- **Pattern Updates**: Arc Core pushes new failure patterns and fixes to Runtime instances
- **Metrics Export**: Local Prometheus endpoint for monitoring Arc Runtime performance
## Installation
```bash
pip install arc-runtime
```
For development:
```bash
git clone https://github.com/arc-computer/runtime.git
cd runtime
pip install -e .
```
## Quick Start
### Zero Configuration
```python
import openai
from runtime import Arc
# Initialize Arc - this automatically patches OpenAI
Arc()
# Use OpenAI as normal - Arc protects your calls
client = openai.OpenAI() # Uses API key from environment
response = client.chat.completions.create(
model="gpt-4.1",
messages=[{"role": "user", "content": "Write a poem about Python"}],
temperature=0.95 # Arc automatically fixes this to 0.7
)
```
### With Telemetry Endpoint
```python
from runtime import Arc
# Connect to your Arc Core instance
arc = Arc(endpoint="grpc://arc.computer:50051")
# All subsequent OpenAI calls are protected and telemetry is streamed
```
## Configuration
Arc Runtime can be configured via environment variables or constructor args:
```python
from runtime import Arc
# Explicit configuration
arc = Arc(
endpoint="grpc://arc.computer:50051",
api_key="arc_key_xxx",
log_level="DEBUG"
)
```
Environment variables:
- `ARC_DISABLE=1` - Disable Arc Runtime completely
- `ARC_ENDPOINT` - gRPC endpoint for telemetry streaming to Arc Core (default: grpc://localhost:50051)
- `ARC_API_KEY` - API key for Arc Core
- `ARC_LOG_LEVEL` - Logging level (default: INFO)
## Metrics
Arc Runtime exposes Prometheus metrics at http://localhost:9090/metrics:
- `arc_requests_intercepted_total` - Total requests intercepted
- `arc_fixes_applied_total` - Total fixes applied
- `arc_pattern_matches_total` - Total pattern matches
- `arc_interception_latency_ms` - Interception overhead histogram
## Custom Patterns
Register custom patterns and fixes:
```python
arc = Arc()
# Register a pattern
arc.register_pattern(
pattern={"model": "gpt-4", "temperature": {">": 0.9}},
fix={"temperature": 0.7}
)
```
## Multi-Agent Pipelines
Track complex multi-agent workflows with automatic context propagation:
```python
from runtime import Arc
import openai
arc = Arc()
client = openai.OpenAI()
# Track a loan underwriting pipeline
with arc.create_multiagent_context(application_id="LOAN-2024-001") as ctx:
# Loan officer agent
response1 = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Analyze loan application"}],
extra_headers={"X-Agent-Name": "loan_officer"}
)
# Credit analyst agent
response2 = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Review credit history"}],
extra_headers={"X-Agent-Name": "credit_analyst"}
)
# Track context handoffs between agents
ctx.add_context_handoff(
from_agent="loan_officer",
to_agent="credit_analyst",
context={"loan_amount": 250000, "initial_assessment": "positive"}
)
# Get pipeline summary
summary = ctx.get_pipeline_summary()
print(f"Agents executed: {summary['agents_executed']}")
print(f"Total latency: {summary['total_latency_ms']}ms")
```
### LangGraph Integration
Automatically track LangGraph workflows:
```python
from runtime import ArcStateGraph
# Use ArcStateGraph instead of StateGraph
workflow = ArcStateGraph()
# Nodes are automatically tracked
workflow.add_node("process_application", process_application_fn)
workflow.add_node("verify_documents", verify_documents_fn)
# Compile and run - Arc tracks everything
app = workflow.compile()
result = app.invoke({"application_id": "APP-123"})
```
## Manual Wrapping
If auto-patching fails, you can explicitly wrap clients:
```python
import openai
from runtime import Arc
arc = Arc()
client = openai.OpenAI()
protected_client = arc.wrap(client)
```
## Default Pattern Fixes
Arc Runtime ships with a built-in pattern for preventing high-temperature hallucinations:
| Pattern | Fix | Rationale |
|---------|-----|-----------|
| GPT-4.1 with temperature > 0.9 | Set temperature to 0.7 | Reduces hallucination risk while maintaining creativity |
### Testing
```bash
# Set your OpenAI API key
export OPENAI_API_KEY="sk-..."
# Run real API tests
python tests/test_real_api.py
```
## Components
- **Interceptors**: Provider-specific hooks (OpenAI, MCP, Anthropic planned)
- **Pattern Registry**: Thread-safe pattern storage and matching
- **Multi-Agent Context**: Pipeline execution tracking with context handoffs
- **MCP Interceptor**: Model Context Protocol monitoring
- **LangGraph Integration**: Automatic workflow tracking
- **Telemetry Client**: OpenTelemetry-compatible async streaming with agent tracing
- **Metrics Server**: Prometheus-compatible metrics endpoint
## Performance
Verified performance characteristics:
- **P99 Interception Overhead**: 0.011ms (requirement: <5ms)
- **Pattern Matching**: <1ms for dictionary lookup
- **Memory Footprint**: <50MB base
- **Thread Safety**: Full concurrent request support
## Troubleshooting
### Arc Runtime is not intercepting calls
1. Ensure Arc is imported before the LLM library:
```python
from runtime import Arc # Import Arc first
Arc()
import openai # Then import OpenAI
```
2. Check if Arc is disabled:
```bash
echo $ARC_DISABLE # Should be empty or "0"
```
3. Enable debug logging:
```bash
export ARC_LOG_LEVEL=DEBUG
```
### Telemetry not streaming
1. Check endpoint connectivity:
```bash
telnet your-arc-endpoint 50051
```
2. Verify gRPC is installed:
```bash
pip install grpcio
```
## License
MIT License - see [LICENSE](LICENSE) for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "arc-runtime",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "ai, llm, openai, anthropic, reliability, monitoring, telemetry, interceptor",
"author": null,
"author_email": "\"Arc Intelligence, Inc.\" <Jarrod@arc.computer>",
"download_url": "https://files.pythonhosted.org/packages/6c/0c/d3aecc32ed5fb702873f68aa5fb7df8673c84b10fc6c4579aae7f1a97df4/arc_runtime-0.1.2.tar.gz",
"platform": null,
"description": "# Arc Runtime\n\n[](https://www.python.org/downloads/)\n[](https://opensource.org/licenses/MIT)\n[](https://badge.fury.io/py/arc-runtime)\n[](docs/performance_report.md)\n[](https://opentelemetry.io/)\n\nArc Runtime is a lightweight Python interceptor that prevents AI agent failures in real-time by applying learned fixes before requests reach the LLM. It's the client-side component of the Arc AI reliability system, designed to work with Arc Core. \n\n## Key Features\n\n- **Zero-config interception** - Just import and it works\n- **Ultra-low latency** - 0.011ms P99 overhead (99.78% better than 5ms requirement)\n- **Thread-safe** - Works seamlessly with async and multi-threaded applications\n- **Pattern matching** - Real-time detection and fixing of known failure patterns\n- **Multi-agent support** - Track complex agent pipelines with context handoffs\n- **MCP interception** - Monitor Model Context Protocol communications\n- **LangGraph integration** - Automatic tracking for LangGraph workflows\n- **OpenTelemetry support** - Full agent telemetry capture (reasoning traces, tool calls, tokens)\n- **Graceful degradation** - Never breaks your application if Arc Core is unreachable\n- **Local metrics** - Prometheus endpoint at http://localhost:9090/metrics\n\n## How It Works\n\nArc Runtime intercepts outgoing LLM API calls and:\n1. Matches requests against known failure patterns (<1ms)\n2. Applies fixes before the request reaches the LLM\n3. Streams telemetry to Arc Core for continuous learning\n4. Exposes metrics for monitoring\n\n## System Architecture\n\nArc Runtime is the client-side component that sits in your application environment:\n\n```mermaid\ngraph TB\n subgraph \"Your Application Environment\"\n App[Your AI Application]\n Arc[Arc Runtime]\n SDK[OpenAI/Anthropic SDK]\n Cache[(Local Cache)]\n \n App --> Arc\n Arc --> Cache\n Cache --> Arc\n Arc --> SDK\n end\n \n SDK --> API[LLM API]\n API --> SDK\n SDK --> Arc\n Arc --> App\n \n subgraph \"Arc Core Service\"\n Collector[gRPC Collector]\n Detector[Failure Detector]\n Registry[Pattern Registry]\n \n Collector --> Detector\n Detector --> Registry\n end\n \n Arc -.-> Collector\n Registry -.-> Cache\n \n style Arc fill:#4CAF50,stroke:#2E7D32,stroke-width:2px\n style App fill:#2196F3,stroke:#1565C0,stroke-width:2px\n style Cache fill:#FFB74D,stroke:#F57C00,stroke-width:2px\n style Collector fill:#E1BEE7,stroke:#9C27B0,stroke-width:2px\n style Registry fill:#FFCDD2,stroke:#D32F2F,stroke-width:2px\n```\n\n**Request Flow:**\n1. Your AI Application makes an API call\n2. Arc Runtime intercepts the request\n3. Checks local cache for matching failure patterns\n4. Applies fixes if patterns match\n5. Forwards the (potentially modified) request to the LLM SDK\n6. SDK sends request to LLM API\n7. Response flows back through Arc Runtime to your application\n8. Arc Runtime asynchronously streams telemetry to Arc Core\n\n**Key Integration Points:**\n- **Telemetry Streaming**: Arc Runtime streams all request/response data to Arc Core via gRPC\n- **Pattern Updates**: Arc Core pushes new failure patterns and fixes to Runtime instances\n- **Metrics Export**: Local Prometheus endpoint for monitoring Arc Runtime performance\n\n## Installation\n\n```bash\npip install arc-runtime\n```\n\nFor development:\n```bash\ngit clone https://github.com/arc-computer/runtime.git\ncd runtime\npip install -e .\n```\n\n## Quick Start\n\n### Zero Configuration\n\n```python\nimport openai\nfrom runtime import Arc\n\n# Initialize Arc - this automatically patches OpenAI\nArc()\n\n# Use OpenAI as normal - Arc protects your calls\nclient = openai.OpenAI() # Uses API key from environment\nresponse = client.chat.completions.create(\n model=\"gpt-4.1\",\n messages=[{\"role\": \"user\", \"content\": \"Write a poem about Python\"}],\n temperature=0.95 # Arc automatically fixes this to 0.7\n)\n```\n\n### With Telemetry Endpoint\n\n```python\nfrom runtime import Arc\n\n# Connect to your Arc Core instance\narc = Arc(endpoint=\"grpc://arc.computer:50051\")\n\n# All subsequent OpenAI calls are protected and telemetry is streamed\n```\n\n## Configuration\n\nArc Runtime can be configured via environment variables or constructor args:\n\n```python\nfrom runtime import Arc\n\n# Explicit configuration\narc = Arc(\n endpoint=\"grpc://arc.computer:50051\",\n api_key=\"arc_key_xxx\",\n log_level=\"DEBUG\"\n)\n```\n\nEnvironment variables:\n- `ARC_DISABLE=1` - Disable Arc Runtime completely\n- `ARC_ENDPOINT` - gRPC endpoint for telemetry streaming to Arc Core (default: grpc://localhost:50051)\n- `ARC_API_KEY` - API key for Arc Core\n- `ARC_LOG_LEVEL` - Logging level (default: INFO)\n\n## Metrics\n\nArc Runtime exposes Prometheus metrics at http://localhost:9090/metrics:\n\n- `arc_requests_intercepted_total` - Total requests intercepted\n- `arc_fixes_applied_total` - Total fixes applied\n- `arc_pattern_matches_total` - Total pattern matches\n- `arc_interception_latency_ms` - Interception overhead histogram\n\n## Custom Patterns\n\nRegister custom patterns and fixes:\n\n```python\narc = Arc()\n\n# Register a pattern\narc.register_pattern(\n pattern={\"model\": \"gpt-4\", \"temperature\": {\">\": 0.9}},\n fix={\"temperature\": 0.7}\n)\n```\n\n## Multi-Agent Pipelines\n\nTrack complex multi-agent workflows with automatic context propagation:\n\n```python\nfrom runtime import Arc\nimport openai\n\narc = Arc()\nclient = openai.OpenAI()\n\n# Track a loan underwriting pipeline\nwith arc.create_multiagent_context(application_id=\"LOAN-2024-001\") as ctx:\n # Loan officer agent\n response1 = client.chat.completions.create(\n model=\"gpt-4\",\n messages=[{\"role\": \"user\", \"content\": \"Analyze loan application\"}],\n extra_headers={\"X-Agent-Name\": \"loan_officer\"}\n )\n \n # Credit analyst agent\n response2 = client.chat.completions.create(\n model=\"gpt-4\",\n messages=[{\"role\": \"user\", \"content\": \"Review credit history\"}],\n extra_headers={\"X-Agent-Name\": \"credit_analyst\"}\n )\n \n # Track context handoffs between agents\n ctx.add_context_handoff(\n from_agent=\"loan_officer\",\n to_agent=\"credit_analyst\",\n context={\"loan_amount\": 250000, \"initial_assessment\": \"positive\"}\n )\n \n # Get pipeline summary\n summary = ctx.get_pipeline_summary()\n print(f\"Agents executed: {summary['agents_executed']}\")\n print(f\"Total latency: {summary['total_latency_ms']}ms\")\n```\n\n### LangGraph Integration\n\nAutomatically track LangGraph workflows:\n\n```python\nfrom runtime import ArcStateGraph\n\n# Use ArcStateGraph instead of StateGraph\nworkflow = ArcStateGraph()\n\n# Nodes are automatically tracked\nworkflow.add_node(\"process_application\", process_application_fn)\nworkflow.add_node(\"verify_documents\", verify_documents_fn)\n\n# Compile and run - Arc tracks everything\napp = workflow.compile()\nresult = app.invoke({\"application_id\": \"APP-123\"})\n```\n\n## Manual Wrapping\n\nIf auto-patching fails, you can explicitly wrap clients:\n\n```python\nimport openai\nfrom runtime import Arc\n\narc = Arc()\nclient = openai.OpenAI()\nprotected_client = arc.wrap(client)\n```\n\n## Default Pattern Fixes\n\nArc Runtime ships with a built-in pattern for preventing high-temperature hallucinations:\n\n| Pattern | Fix | Rationale |\n|---------|-----|-----------|\n| GPT-4.1 with temperature > 0.9 | Set temperature to 0.7 | Reduces hallucination risk while maintaining creativity |\n\n\n### Testing\n\n```bash\n# Set your OpenAI API key\nexport OPENAI_API_KEY=\"sk-...\"\n\n# Run real API tests\npython tests/test_real_api.py\n```\n\n## Components\n\n- **Interceptors**: Provider-specific hooks (OpenAI, MCP, Anthropic planned)\n- **Pattern Registry**: Thread-safe pattern storage and matching\n- **Multi-Agent Context**: Pipeline execution tracking with context handoffs\n- **MCP Interceptor**: Model Context Protocol monitoring\n- **LangGraph Integration**: Automatic workflow tracking\n- **Telemetry Client**: OpenTelemetry-compatible async streaming with agent tracing\n- **Metrics Server**: Prometheus-compatible metrics endpoint\n\n## Performance\n\nVerified performance characteristics:\n- **P99 Interception Overhead**: 0.011ms (requirement: <5ms)\n- **Pattern Matching**: <1ms for dictionary lookup\n- **Memory Footprint**: <50MB base\n- **Thread Safety**: Full concurrent request support\n\n## Troubleshooting\n\n### Arc Runtime is not intercepting calls\n\n1. Ensure Arc is imported before the LLM library:\n ```python\n from runtime import Arc # Import Arc first\n Arc()\n import openai # Then import OpenAI\n ```\n\n2. Check if Arc is disabled:\n ```bash\n echo $ARC_DISABLE # Should be empty or \"0\"\n ```\n\n3. Enable debug logging:\n ```bash\n export ARC_LOG_LEVEL=DEBUG\n ```\n\n### Telemetry not streaming\n\n1. Check endpoint connectivity:\n ```bash\n telnet your-arc-endpoint 50051\n ```\n\n2. Verify gRPC is installed:\n ```bash\n pip install grpcio\n ```\n\n## License\n\nMIT License - see [LICENSE](LICENSE) for details.\n",
"bugtrack_url": null,
"license": null,
"summary": "Lightweight Python interceptor that prevents AI agent failures in real-time",
"version": "0.1.2",
"project_urls": {
"Bug Tracker": "https://github.com/arc-computer/runtime/issues",
"Documentation": "https://github.com/arc-computer/runtime#readme",
"Homepage": "https://github.com/arc-computer/runtime",
"Source Code": "https://github.com/arc-computer/runtime"
},
"split_keywords": [
"ai",
" llm",
" openai",
" anthropic",
" reliability",
" monitoring",
" telemetry",
" interceptor"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "9ec8a79de7fa4111277e6424de8c6362b8707cce25684f6c7b3800649e9e4e00",
"md5": "dd73112a89f5f5ffdea6a10f7280b3c3",
"sha256": "500a05fc97b599e4d1bcf6b2b65b290abc1d5e971fd4d64b74df376098ff2f35"
},
"downloads": -1,
"filename": "arc_runtime-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "dd73112a89f5f5ffdea6a10f7280b3c3",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 37116,
"upload_time": "2025-07-08T16:01:45",
"upload_time_iso_8601": "2025-07-08T16:01:45.401796Z",
"url": "https://files.pythonhosted.org/packages/9e/c8/a79de7fa4111277e6424de8c6362b8707cce25684f6c7b3800649e9e4e00/arc_runtime-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "6c0cd3aecc32ed5fb702873f68aa5fb7df8673c84b10fc6c4579aae7f1a97df4",
"md5": "cdc04a1c8ce155e21544f588d9bbc82c",
"sha256": "d9c6170033e4abea85702896ab8524b0b55a6dfb9d6b35d49d6b14b534fc4294"
},
"downloads": -1,
"filename": "arc_runtime-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "cdc04a1c8ce155e21544f588d9bbc82c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 53189,
"upload_time": "2025-07-08T16:01:46",
"upload_time_iso_8601": "2025-07-08T16:01:46.673894Z",
"url": "https://files.pythonhosted.org/packages/6c/0c/d3aecc32ed5fb702873f68aa5fb7df8673c84b10fc6c4579aae7f1a97df4/arc_runtime-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-08 16:01:46",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "arc-computer",
"github_project": "runtime",
"github_not_found": true,
"lcname": "arc-runtime"
}
JSON
