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 (Dict[str, Any] | None)

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)

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)

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_id)

Find agents by organization.

Parameters:

organization_id (str) – ID 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

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.

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

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

Return type:

bool

Returns:

True if the user owns the agent, False otherwise

class AgentRegistration(agent_id, organization_id, agent_type, interaction_modes, capabilities, identity, owner_id=None, payment_address=None, metadata=<factory>)

Bases: object

Registration information for an agent.

This class stores the registration information for an agent, including its identity, capabilities, and interaction modes.

Parameters:

• agent_id (str)

• organization_id (str | None)

• agent_type (AgentType)

• interaction_modes (list[InteractionMode])

• capabilities (list[Capability])

• identity (AgentIdentity)

• owner_id (str | None)

• payment_address (str | None)

• metadata (Dict)

agent_id

Unique identifier for the agent

organization_id

ID of the organization the agent belongs to

agent_type

Type of agent (human, AI)

interaction_modes

Supported interaction modes

capabilities

List of agent capabilities

identity

Agent’s decentralized identity

owner_id

ID of the agent’s owner

payment_address

Agent’s primary wallet address for receiving payments

metadata

Additional information about the agent

owner_id: Optional[str] = None

payment_address: Optional[str] = None

agent_id: str

organization_id: Optional[str]

agent_type: AgentType

interaction_modes: list[InteractionMode]

capabilities: list[Capability]

identity: AgentIdentity

metadata: Dict

class CapabilityDiscoveryService(vector_store_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 vector stores.

Parameters:

vector_store_config (Dict[str, Any])

clear_agent_embeddings_cache(agent_id)

Clear the embeddings cache for a specific agent.

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)

Find agents by capability description using semantic search.

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) For negative scores (distance metrics), this becomes a maximum threshold (lower absolute value means more similar)

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.

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

async load_vector_store(path, embeddings_model=None)

Load the vector store from disk.

Parameters:

• path (str) – Directory path to load the vector store from

• embeddings_model – Optional embeddings model to use (if None, uses the current one)

Return type:

bool

Returns:

True if successful, False otherwise

async precompute_all_capability_embeddings(agent_registrations)

Precompute embeddings for all existing capabilities.

Parameters:

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

Return type:

None

async save_vector_store(path)

Save the vector store to disk for faster loading in the future.

Parameters:

path (str) – Directory path to save the vector store

Return type:

bool

Returns:

True if successful, False otherwise

async update_capability_embeddings_cache(registration)

Update capability embeddings for a registration.

Parameters:

registration (AgentRegistration) – Registration information for the agent

Return type:

None

Submodules

• agentconnect.core.registry.capability_discovery module
  • custom_showwarning()
  • check_semantic_search_requirements()
  • calculate_similarity()
  • cosine_similarity()
  • CapabilityDiscoveryService
    • CapabilityDiscoveryService.initialize_embeddings_model()
    • CapabilityDiscoveryService.update_capability_embeddings_cache()
    • CapabilityDiscoveryService.clear_agent_embeddings_cache()
    • CapabilityDiscoveryService.precompute_all_capability_embeddings()
    • CapabilityDiscoveryService.find_by_capability_name()
    • CapabilityDiscoveryService.find_by_capability_semantic()
    • CapabilityDiscoveryService.save_vector_store()
    • CapabilityDiscoveryService.load_vector_store()
• agentconnect.core.registry.identity_verification module
  • verify_agent_identity()
  • verify_ethereum_did()
  • verify_key_did()
• agentconnect.core.registry.registration module
  • AgentRegistration
    • AgentRegistration.agent_id
    • AgentRegistration.organization_id
    • AgentRegistration.agent_type
    • AgentRegistration.interaction_modes
    • AgentRegistration.capabilities
    • AgentRegistration.identity
    • AgentRegistration.owner_id
    • AgentRegistration.payment_address
    • AgentRegistration.metadata
    • AgentRegistration.agent_id
    • AgentRegistration.organization_id
    • AgentRegistration.agent_type
    • AgentRegistration.interaction_modes
    • AgentRegistration.capabilities
    • AgentRegistration.identity
    • AgentRegistration.owner_id
    • AgentRegistration.payment_address
    • AgentRegistration.metadata
• agentconnect.core.registry.registry_base module
  • AgentRegistry
    • AgentRegistry.register()
    • AgentRegistry.unregister()
    • AgentRegistry.get_by_capability()
    • AgentRegistry.get_by_capability_semantic()
    • AgentRegistry.get_all_capabilities()
    • AgentRegistry.get_all_agents()
    • AgentRegistry.get_agent_type()
    • AgentRegistry.get_by_interaction_mode()
    • AgentRegistry.get_registration()
    • AgentRegistry.get_by_organization()
    • AgentRegistry.get_verified_agents()
    • AgentRegistry.verify_agent()
    • AgentRegistry.update_registration()
    • AgentRegistry.get_by_owner()
    • AgentRegistry.verify_owner()