synqed 1.0.5

Synqed - A wrapper around A2A for simplified multi-agent systems interaction and communication

Synqed Python API library

Synqed enables true AI-to-AI interaction.

Agents can talk to each other, collaborate, coordinate, delegate tasks, and solve problems together—letting you build actual multi-agent systems.

All seamless. All autonomous.

Synqed also lets agents from any provider—OpenAI, Anthropic, Google, or local models—communicate as part of the same system.

Documentation

For full API documentation, see here

Installation

# install from PyPI
pip install synqed

Synqed works with the following LLM providers. Install your preferred provider:

pip install openai                  # For OpenAI (GPT-4, GPT-4o, etc.)
pip install anthropic               # For Anthropic (Claude)
pip install google-generativeai     # For Google (Gemini)

Usage

Quick Start: Your First Agent

Here's the fastest way to get started:

Create a file my_agent.py:

import asyncio
import os
import synqed

async def agent_logic(context):
    """Your agent's brain - this is where the magic happens."""
    user_message = context.get_user_input()
    
    # Use any LLM you want
    from openai import AsyncOpenAI
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": user_message}
        ]
    )
    
    return response.choices[0].message.content

async def main():
    # Create your agent
    agent = synqed.Agent(
        name="MyFirstAgent",
        description="A helpful AI assistant",
        skills=["general_assistance", "question_answering"],
        executor=agent_logic
    )
    
    # Start the server
    server = synqed.AgentServer(agent, port=8000)
    print(f"Agent running at {agent.url}")
    await server.start()

if __name__ == "__main__":
    asyncio.run(main())

Step 2: Connect a Client

Create a file client.py:

import asyncio
import synqed

async def main():
    async with synqed.Client("http://localhost:8000") as client:
        # Option 1: Simple request-response
        response = await client.ask("What's the weather like?")
        print(f"Agent: {response}")
        
        # Option 2: Streaming response (like ChatGPT typing)
        print("Streaming: ", end="")
        async for chunk in client.stream("Tell me a joke"):
            print(chunk, end="", flush=True)
        print()

if __name__ == "__main__":
    asyncio.run(main())

Step 3: Run It

# Terminal 1 - Start your agent
python my_agent.py

# Terminal 2 - Connect your client
python client.py

Congratulations! You just built and deployed your first AI agent.

---

Understanding Executor Functions

The executor is where you define your agent's behavior. It receives a context object and returns a response:

async def agent_logic(context):
    """
    Args:
        context: RequestContext with methods:
            - get_user_input() → str: User's message
            - get_task() → Task: Full task object
            - get_message() → Message: Full message object
    
    Returns:
        str or Message: Agent's response
    """
    user_message = context.get_user_input()
    
    # Implement any logic:
    # - Call LLMs (OpenAI, Anthropic, Google)
    # - Query databases
    # - Call external APIs
    # - Delegate to other agents
    
    return "Agent response"

Client Configuration

The client allows your agents to interact with other agents.

import synqed

# Default configuration
client = synqed.Client("http://localhost:8000")

# Custom timeout
client = synqed.Client(
    agent_url="http://localhost:8000",
    timeout=120.0  # 2 minutes (default is 60)
)

# Disable streaming
client = synqed.Client(
    agent_url="http://localhost:8000",
    streaming=False
)

# Override per-request
async with synqed.Client("http://localhost:8000") as client:
    response = await client.with_options(timeout=30.0).ask("Quick question")

---

Agent Collaboration with Orchestrator

The Orchestrator uses an LLM to analyze tasks and intelligently route them to the most suitable agents.

Basic Orchestration

import synqed
import os

# Create orchestrator with LLM-powered routing
orchestrator = synqed.Orchestrator(
    provider=synqed.LLMProvider.OPENAI,
    api_key=os.environ.get("OPENAI_API_KEY"),
    model="gpt-4o"
)

# Register your specialized agents to the orchestrator
orchestrator.register_agent(research_agent.card, "http://localhost:8001")
orchestrator.register_agent(coding_agent.card, "http://localhost:8002")
orchestrator.register_agent(writing_agent.card, "http://localhost:8003")

# Orchestrator automatically selects the best agent(s) for the task
result = await orchestrator.orchestrate(
    "Research recent AI developments and write a technical summary"
)

print(f"Selected: {result.selected_agents[0].agent_name}")
print(f"Confidence: {result.selected_agents[0].confidence:.0%}")
print(f"Reasoning: {result.selected_agents[0].reasoning}")

Supported LLM Providers

import synqed

# OpenAI
synqed.Orchestrator(
    provider=synqed.LLMProvider.OPENAI,
    api_key=os.environ.get("OPENAI_API_KEY"),
    model="model-here" 
)

# Anthropic
synqed.Orchestrator(
    provider=synqed.LLMProvider.ANTHROPIC,
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    model="model-here"
)

# Google
synqed.Orchestrator(
    provider=synqed.LLMProvider.GOOGLE,
    api_key=os.environ.get("GOOGLE_API_KEY"),
    model="model-here"
)

Orchestration Configuration

import synqed

orchestrator = synqed.Orchestrator(
    provider=synqed.LLMProvider.OPENAI,
    api_key=os.environ.get("OPENAI_API_KEY"),
    model="gpt-4o",
    temperature=0.7,     # Creativity level (0.0 - 1.0)
    max_tokens=2000      # Maximum response length
)

---

Multi-Agent Delegation

The TaskDelegator coordinates multiple agents working together on complex tasks:

import synqed
import os

# Create orchestrator for intelligent routing
orchestrator = synqed.Orchestrator(
    provider=synqed.LLMProvider.OPENAI,
    api_key=os.environ.get("OPENAI_API_KEY"),
    model="gpt-4o"
)

# Create delegator
delegator = synqed.TaskDelegator(orchestrator=orchestrator)

# Register specialized agents (local or remote)
delegator.register_agent(agent=research_agent)
delegator.register_agent(agent=coding_agent)
delegator.register_agent(agent=writing_agent)

# Agents automatically collaborate on complex tasks
result = await delegator.submit_task(
    "Research microservices patterns and write implementation guide"
)

Remote Agent Registration

Register agents running anywhere:

# Register remote agent
delegator.register_agent(
    agent_url="https://specialist-agent.example.com",
    agent_card=agent_card  # Optional pre-loaded card
)

---

Complete Examples

Ready to dive deeper? Check out the complete, runnable examples here

---

Copyright © 2025 Synq Team. All rights reserved.