agentconnect.clients package

AgentConnect Clients Package

This package provides client implementations for interacting with AgentConnect services, offering both low-level API access and high-level abstractions for remote agent management.

Available Clients:

  • RegistryAPIClient: HTTP client for AgentConnect Registry API Server

class RegistryAPIClient(base_url=None, timeout=None, connect_timeout=None, read_timeout=None, pool_timeout=None, max_connections=None, max_keepalive_connections=None)

Bases: object

Client for interacting with the AgentConnect Registry API Server. This client mimics the interface of agentconnect.core.registry.AgentRegistry.

Quickstart Example:
import asyncio
from agentconnect.clients import RegistryAPIClient

async def main():
    async with RegistryAPIClient() as client:
        # Get all registered agents
        agents = await client.get_all_agents()
        print(f"Found {len(agents)} agents")

        # Search for agents by capability
        results = await client.get_by_capability_semantic(
            capability_description="data analysis",
            limit=5
        )
        for agent, score in results:
            print(f"{agent.name}: {score:.3f}")

asyncio.run(main())
Parameters:

  • base_url (str | None)

  • timeout (float | None)

  • connect_timeout (float | None)

  • read_timeout (float | None)

  • pool_timeout (float | None)

  • max_connections (int | None)

  • max_keepalive_connections (int | None)

async close()

Closes the underlying HTTPX client. Should be called on cleanup.

async ensure_initialized()

Mimics AgentRegistry’s ensure_initialized. For the client, this is a no-op.

async get_agent_type(agent_id)

Get the type of an agent.

Return type:

Optional[AgentType]

Parameters:

agent_id (str)

async get_all_agents()

Get a list of all agents registered in the system.

Return type:

List[AgentRegistration]

async get_all_capabilities()

Get a list of all unique capability names registered in the system.

Return type:

List[str]

async get_by_capability(capability_name, limit=10, similarity_threshold=0.1)

Find agents by capability name (exact match).

Return type:

List[AgentRegistration]

Parameters:

  • capability_name (str)

  • limit (int)

  • similarity_threshold (float)

async get_by_capability_semantic(capability_description, limit=10, similarity_threshold=0.1, filters=None)

Find agents by capability description using semantic search.

Return type:

List[Tuple[AgentRegistration, float]]

Parameters:
async get_by_interaction_mode(mode)

Find agents by interaction mode.

Return type:

List[AgentRegistration]

Parameters:

mode (InteractionMode)

async get_by_organization(organization)

Find agents by organization.

Return type:

List[AgentRegistration]

Parameters:

organization (str)

async get_by_owner(owner_id)

Find agents by owner (developer).

Return type:

List[AgentRegistration]

Parameters:

owner_id (str)

async get_registration(agent_id)

Get agent registration details.

Return type:

Optional[AgentRegistration]

Parameters:

agent_id (str)

async get_verified_agents()

Get all verified agents.

Return type:

List[AgentRegistration]

async register(registration)

Register a new agent.

Return type:

bool

Parameters:

registration (AgentRegistration)

async unregister(agent_id)

Remove agent from registry.

Return type:

bool

Parameters:

agent_id (str)

async update_registration(agent_id, updates)

Update agent registration details.

Return type:

Optional[AgentRegistration]

Parameters:
async verify_agent(agent_id)

Verify an agent’s identity (triggers verification process on server).

Return type:

bool

Parameters:

agent_id (str)

async verify_owner(agent_id, owner_id)

Verify if a user owns an agent (developer).

Return type:

bool

Parameters:

  • agent_id (str)

  • owner_id (str)

Submodules