agentconnect.core.registry package

Agent registry for the AgentConnect framework.

This module provides the AgentRegistry class for agent registration, discovery, and capability matching, as well as the AgentRegistration dataclass for storing agent registration information.

class AgentRegistry(vector_search_config=None)

Bases: object

Central registry for agent registration and discovery.

This class provides methods for registering agents, discovering agents by capability, and verifying agent identities.

Parameters:

vector_search_config (VectorSearchSettings | Dict[str, Any] | None)

async ensure_initialized()

Wait until the core registry initialization is complete.

async get_agent_type(agent_id)

Get the type of an agent.

Parameters:

agent_id (str) – ID of the agent

Return type:

AgentType

Returns:

Type of the agent

Raises:

KeyError – If the agent is not found

async get_all_agents()

Get a list of all agents registered in the system.

Return type:

List[AgentRegistration]

Returns:

List of all agent registrations

async get_all_capabilities()

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

Return type:

List[str]

Returns:

List of all capability names

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

Find agents by capability name.

Parameters:

  • capability_name (str) – Name of the capability to search for

  • limit (int) – Maximum number of results to return (default: 10)

  • similarity_threshold (float) – Minimum similarity score for semantic fallback search (default: 0.1)

Return type:

List[AgentRegistration]

Returns:

List of agent registrations with the specified capability

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

Find agents by capability description using semantic search.

Parameters:

  • capability_description (str) – Description of the capability to search for

  • limit (int) – Maximum number of results to return (default: 10)

  • similarity_threshold (float) – Minimum similarity score to include in results (default: 0.1)

  • filters (Optional[Dict[str, List[str]]]) – Optional dictionary for filtering. Keys can include “tags”, “organization”, “developer”, “default_input_modes”, “default_output_modes”, “auth_schemes”. Values are lists of strings to match for the respective key.

Return type:

List[Tuple[AgentRegistration, float]]

Returns:

List of tuples containing agent registrations and similarity scores

async get_by_interaction_mode(mode)

Find agents by interaction mode.

Parameters:

mode (InteractionMode) – Interaction mode to search for

Return type:

List[AgentRegistration]

Returns:

List of agent registrations with the specified interaction mode

async get_by_organization(organization)

Find agents by organization.

Parameters:

organization (str) – ID/name of the organization

Return type:

List[AgentRegistration]

Returns:

List of agent registrations in the specified organization

async get_by_owner(owner_id)

Find agents by owner.

Parameters:

owner_id (str) – ID of the owner (now using developer field)

Return type:

List[AgentRegistration]

Returns:

List of agent registrations owned by the specified owner

async get_registration(agent_id)

Get agent registration details.

Parameters:

agent_id (str) – ID of the agent

Return type:

Optional[AgentRegistration]

Returns:

Agent registration if found, None otherwise

async get_verified_agents()

Get all verified agents.

Return type:

List[AgentRegistration]

Returns:

List of verified agent registrations

async register(registration)

Register a new agent with verification. Waits for initialization first.

Parameters:

registration (AgentRegistration) – Registration information for the agent

Return type:

bool

Returns:

True if registration was successful, False otherwise

async unregister(agent_id)

Remove agent from registry.

Parameters:

agent_id (str) – ID of the agent to unregister

Return type:

bool

Returns:

True if unregistration was successful, False otherwise

async update_registration(agent_id, updates)

Update agent registration details.

Parameters:

  • agent_id (str) – ID of the agent to update

  • updates (Dict) – Dictionary of updates to apply

Return type:

Optional[AgentRegistration]

Returns:

Updated agent registration if successful, None otherwise

property vector_search_settings: VectorSearchSettings

Get the vector search settings as a Pydantic model.

async verify_agent(agent_id)

Verify an agent’s identity.

Parameters:

agent_id (str) – ID of the agent to verify

Return type:

bool

Returns:

True if verification was successful, False otherwise

async verify_owner(agent_id, owner_id)

Verify if a user owns an agent.

Parameters:

  • agent_id (str) – ID of the agent

  • owner_id (str) – ID of the owner (now using developer field)

Return type:

bool

Returns:

True if the user owns the agent, False otherwise

class AgentRegistration(**data)

Bases: BaseModel

Registration information for an agent.

This class stores the complete registration information for an agent, including its identity, capabilities, skills, and metadata needed for discovery and interaction.

Parameters:
agent_id

Unique identifier for the agent

agent_type

Type of agent (human, AI)

interaction_modes

Supported interaction modes

identity

Agent’s decentralized identity

name

Name of the agent

summary

Brief summary of the agent’s purpose

description

Detailed description of the agent

version

Version of the agent

documentation_url

URL to the agent’s documentation

organization

Organization or entity providing the agent (e.g., ‘Acme Corp’, ‘did:org:123’). Using a verifiable ID is recommended for robustness.

developer

Individual or team that developed the agent (e.g., ‘Alice’, ‘did:person:abc’). Using a verifiable ID is recommended.

url

Endpoint URL for the agent

auth_schemes

List of supported authentication schemes

default_input_modes

List of supported input modes

default_output_modes

List of supported output modes

capabilities

List of capabilities the agent provides

skills

List of skills the agent possesses

examples

Example inputs/outputs or use cases

tags

Keywords for filtering

payment_address

Agent’s primary wallet address for receiving payments

custom_metadata

Additional custom metadata about the agent

registered_at

When the agent was registered

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

agent_id: str
agent_type: AgentType
interaction_modes: list[InteractionMode]
identity: AgentIdentity
name: Optional[str]
summary: Optional[str]
description: Optional[str]
version: Optional[str]
documentation_url: Optional[str]
organization: Optional[str]
developer: Optional[str]
url: Optional[str]
auth_schemes: List[str]
default_input_modes: List[str]
default_output_modes: List[str]
capabilities: List[Capability]
skills: List[Skill]
examples: List[str]
tags: List[str]
payment_address: Optional[str]
custom_metadata: Dict[str, Any]
registered_at: datetime
class CapabilityDiscoveryService(vector_search_config=None)

Bases: object

Service for discovering agent capabilities through various search methods.

This class provides methods for finding agents based on their capabilities, including exact string matching and semantic search using Qdrant vector database.

Parameters:

vector_search_config (Optional[Union[VectorSearchSettings, Dict[str, Any]]])

COLLECTION_NAME = 'agent_capabilities'
async clear_agent_embeddings_cache(agent_id)

Clear the embeddings cache for a specific agent from Qdrant.

Parameters:

agent_id (str) – ID of the agent to clear cache for

Return type:

None

async find_by_capability_name(capability_name, agent_registrations, capabilities_index, limit=10, similarity_threshold=0.1)

Find agents by capability name (simple string matching).

Parameters:

  • capability_name (str) – Name of the capability to search for

  • agent_registrations (Dict[str, AgentRegistration]) – Dictionary of agent registrations

  • capabilities_index (Dict[str, Set[str]]) – Index of agent capabilities

  • limit (int) – Maximum number of results to return (default: 10)

  • similarity_threshold (float) – Minimum similarity score to include in results (default: 0.1)

Return type:

List[AgentRegistration]

Returns:

List of agent registrations with the specified capability

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

Find agents by capability description using semantic search with metadata filtering.

Parameters:

  • capability_description (str) – Description of the capability to search for

  • agent_registrations (Dict[str, AgentRegistration]) – Dictionary of agent registrations

  • limit (int) – Maximum number of results to return (default: 10)

  • similarity_threshold (float) – Minimum similarity score to include in results (default: 0.1)

  • filters (Optional[Dict[str, List[str]]]) – Optional dictionary for filtering. Keys can include “tags”, “organization”, “developer”, “default_input_modes”, “default_output_modes”, “auth_schemes”. Values are lists of strings to match for the respective key.

Return type:

List[Tuple[AgentRegistration, float]]

Returns:

List of tuples containing agent registrations and similarity scores

async initialize_embeddings_model()

Initialize the embeddings model for semantic search and Qdrant client.

This should be called after agents have been registered to precompute embeddings for all existing capabilities.

async precompute_all_capability_embeddings(agent_registrations)

Precompute embeddings for all existing capabilities and store in Qdrant.

Parameters:

agent_registrations (Dict[str, AgentRegistration]) – Dictionary of agent registrations

Return type:

None

async update_capability_embeddings_cache(registration)

Update capability embeddings for a registration in Qdrant.

Parameters:

registration (AgentRegistration) – Registration information for the agent

Return type:

None

Subpackages

Submodules

On this page
Edit on GitHub