JarvisCore

JarvisCore is an open-source framework designed to orchestrate teams of agents to perform complex tasks from a single, auditable control plane.

Scroll to explore

pip install jarviscore-framework

Developer Center

View Latest Updates

Loading updates…

PLATFORM CAPABILITIES

Primitives, tools, and protocols to orchestrate deterministic AI.

Everything you need to build, deploy, and scale enterprise-grade multi-agent systems without the underlying infrastructure complexities.

Two execution models

AutoAgent handles the full reasoning loop. CustomAgent exposes each message handler so you decide what runs and when.

Learn about agent profiles

import asyncio
from jarviscore import Mesh, AutoAgent


class ResearcherAgent(AutoAgent):
    name = "Researcher"
    role = "researcher"
    system_prompt = "You are a rigorous research analyst."


async def main():
    mesh = Mesh()
    mesh.add(ResearcherAgent)
    await mesh.start()
    result = await mesh.run_task(
        agent="researcher",
        task="What are the key architectural trade-offs in multi-agent systems?",
    )
    print(result)


asyncio.run(main())

Four-tier memory

Four tiers compose behind a single object. Each tier activates based on what you pass to UnifiedMemory at startup.

Explore the memory model

from jarviscore.memory import UnifiedMemory

mem = UnifiedMemory(
    workflow_id="wf-abc123",
    step_id="step-1",
    agent_id="researcher",
    blob_storage=blob_storage,
)

await mem.log_turn(
    turn_id="t1",
    thought="Identifying relevant data sources.",
    action="http_get",
    result="200 OK - 42 records retrieved",
    tokens=1240,
)

Peer-to-peer mesh

One flag. Agents discover each other via SWIM gossip over ZMQ. Works the same on one process or ten machines.

Read the P2P docs

# Enable by setting P2P_ENABLED=true in your environment.

P2P_ENABLED=true

P2P_ENABLED=true
JC_SWIM_HOST=0.0.0.0
JC_SWIM_PORT=7947
JC_SEED_NODES=10.0.0.1:7946

analyst = self.peers.get_peer(role="analyst")

workers = self.peers.discover(
    capability="data-analysis",
    strategy="round_robin",  # first | random | round_robin | least_recent
)

worker = self.peers.discover_one(
    capability="processing",
    strategy="least_recent",
)

Add custom integrations

Write a plain Python function. Drop it in the atoms directory. Every agent on the Mesh picks it up automatically.

Build your first Atom

def crm_get_contact(auth_info: dict, contact_id: str) -> dict:
    """
    Fetch a contact from the internal CRM.

    Args:
        auth_info: Injected by Nexus. Contains the API token.
        contact_id: CRM contact ID.

    Returns:
        {"id": str, "name": str, "email": str}
    """
    import httpx
    headers = {"Authorization": f"Bearer {auth_info['token']}"}
    r = httpx.get(
        f"https://crm.internal/contacts/{contact_id}",
        headers=headers,
    )
    return r.json()

Nexus credential layer

Set up Nexus auth

jarviscore nexus register github     --client-id=YOUR_GITHUB_CLIENT_ID     --client-secret=YOUR_GITHUB_CLIENT_SECRET

jarviscore nexus register stripe --api-key=sk_live_...
jarviscore nexus register airtable --api-key=patXXXXXXXX

jarviscore nexus list

Full-stack tracing

Every turn, tool call, and LLM request is captured automatically. Redis PubSub for live streams, JSONL for compliance, Prometheus for SLO dashboards.

Configure observability

from jarviscore.telemetry import TraceManager

tracer = TraceManager(
    workflow_id="wf-abc123",
    step_id="step-001",
    redis_store=redis_store,
    trace_dir="traces",
)

tracer.log_tool_start("slack_send_message", params={"channel": "#alerts"})
tracer.log_tool_result("slack_send_message", result="ok")
tracer.log_thinking("Evaluating whether output meets acceptance criteria")
tracer.log_event("custom_event", data={"detail": "something happened"})

Human-in-the-loop

Pause on irreversible actions and route to a human inbox. Resumption is automatic once resolved.

Learn about HITL escalation

jarviscore.dev/hitl

item_id = self.hitl.request(
    title="Approve investor deck before sending",
    content="The deck is ready for Q2 distribution. Approve to proceed.",
    urgency="high",
    category="critical_action",
    context={"file": "output/q2_deck.pdf", "recipients": 47},
)

resolution = await self.hitl.wait(item_id, timeout=3600)

if resolution.is_approved:
    await self.send_deck()
else:
    return {"status": "cancelled", "reason": resolution.reason}