Skip to content

A2A Registry API Reference

Overview

The A2A Registry provides a multi-protocol API for agent registration, discovery, and management. This reference covers the primary interaction methods across JSON-RPC 2.0, REST, and GraphQL protocols.

Authentication & Security

Trust Levels

  • Development Mode: Open access, for local development
  • Production Mode: Strict authentication and authorization
  • JWT-based authentication
  • Role-based access control (RBAC)
  • Configurable trust levels for agent registration

Authentication Methods

  1. JWT Token
  2. API Key
  3. OAuth 2.0

Core API Methods

Agent Registration

JSON-RPC 2.0

{
  "jsonrpc": "2.0",
  "method": "register_agent",
  "params": {
    "agent_card": {
      "name": "example-agent",
      "description": "Agent description",
      "version": "1.0.0",
      "protocol_version": "0.3.0",
      "preferred_transport": "JSONRPC",
      "skills": [
        {
          "id": "skill_id",
          "description": "Skill description"
        }
      ]
    }
  },
  "id": 1
}

REST Endpoint

POST /api/v1/agents
Content-Type: application/json

{
  "name": "example-agent",
  ...
}

GraphQL Mutation

mutation {
  registerAgent(
    agentCard: {
      name: "example-agent"
      # ... other fields
    }
  ) {
    id
    name
    registrationTimestamp
  }
}

Agent Discovery

Search Methods

  • By skill
  • By name/description
  • By trust level
  • By protocol version
{
  "jsonrpc": "2.0",
  "method": "search_agents",
  "params": {
    "skills": ["weather_forecast"],
    "protocol_version": "0.3.0"
  },
  "id": 2
}

GraphQL Query

query {
  searchAgents(
    skills: ["weather_forecast"]
    protocolVersion: "0.3.0"
  ) {
    agents {
      name
      skills {
        id
        description
      }
    }
  }
}

Auto-Generated Module Documentation

Server Module

A2A Registry server using FastAPI and FastA2A schemas with dual transport support.

GRAPHQL_AVAILABLE = False module-attribute

logger = logging.getLogger(__name__) module-attribute

AgentSearchRequest

Bases: BaseModel

Request to search for agents.

query instance-attribute

RegisterAgentRequest

Bases: BaseModel

Request to register an agent.

agent_card instance-attribute

create_app()

Create FastAPI application for A2A Registry.

Source code in src/a2a_registry/server.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def create_app() -> FastAPI:
    """Create FastAPI application for A2A Registry."""
    app = FastAPI(
        title="A2A Registry",
        description="Agent-to-Agent Registry Service with GraphQL",
        version=__version__,
    )

    # Try to initialize GraphQL extension storage and setup GraphQL API if available
    try:
        from .graphql.app import setup_graphql_with_fastapi
        from .graphql.storage import InMemoryExtensionStorage

        extension_storage = InMemoryExtensionStorage()
        setup_graphql_with_fastapi(app, extension_storage, storage)
        GRAPHQL_AVAILABLE = True
        logger.info("GraphQL support enabled")
    except Exception as e:
        logger.info(f"GraphQL support disabled - only REST API available: {e}")
        GRAPHQL_AVAILABLE = False

    @app.post("/agents", response_model=dict[str, Any])
    async def register_agent(request: RegisterAgentRequest) -> dict[str, Any]:
        """Register an agent in the registry."""
        try:
            # AgentCard is a TypedDict, so we can use the dict directly
            # but we should validate required fields
            agent_card_dict = request.agent_card

            # Validate required fields for AgentCard
            required_fields = [
                "name",
                "description",
                "url",
                "version",
                "protocol_version",
            ]
            for field in required_fields:
                if field not in agent_card_dict:
                    raise ValueError(f"Missing required field: {field}")

            # Set default transport to JSONRPC per A2A specification
            if "preferred_transport" not in agent_card_dict:
                agent_card_dict["preferred_transport"] = "JSONRPC"

            # Cast to AgentCard type for type safety
            agent_card: AgentCard = agent_card_dict  # type: ignore
            success = await storage.register_agent(agent_card)

            if success:
                # Extract and update agent extensions
                agent_id = agent_card["name"]
                extensions = []

                # Get extensions from agent card capabilities
                capabilities = agent_card.get("capabilities", {})
                if isinstance(capabilities, dict):
                    agent_extensions = capabilities.get("extensions", [])
                    if isinstance(agent_extensions, list):
                        extensions = [
                            {
                                "uri": ext.get("uri", ""),
                                "description": ext.get("description", ""),
                                "required": ext.get("required", False),
                                "params": ext.get("params", {}),
                            }
                            for ext in agent_extensions
                            if isinstance(ext, dict) and ext.get("uri")
                        ]

                # Update agent extensions in storage
                await storage.update_agent_extensions(agent_id, extensions)

                return {
                    "success": True,
                    "agent_id": agent_id,
                    "message": "Agent registered successfully",
                    "extensions_processed": len(extensions),
                }
            else:
                raise HTTPException(status_code=400, detail="Failed to register agent")

        except Exception as e:
            logger.error(f"Error registering agent: {e}")
            raise HTTPException(status_code=400, detail=str(e)) from e

    @app.get("/agents/{agent_id}", response_model=dict[str, Any])
    async def get_agent(agent_id: str) -> dict[str, Any]:
        """Get an agent by ID."""
        agent_card = await storage.get_agent(agent_id)
        if agent_card:
            return {"agent_card": dict(agent_card)}
        else:
            raise HTTPException(status_code=404, detail="Agent not found")

    @app.get("/agents", response_model=dict[str, Any])
    async def list_agents() -> dict[str, Any]:
        """List all registered agents."""
        agents = await storage.list_agents()
        return {"agents": [dict(agent) for agent in agents], "count": len(agents)}

    @app.delete("/agents/{agent_id}", response_model=dict[str, Any])
    async def unregister_agent(agent_id: str) -> dict[str, Any]:
        """Unregister an agent."""
        # First remove agent from extensions
        extensions_removed = await storage.remove_agent_from_extensions(agent_id)

        # Then remove the agent
        success = await storage.unregister_agent(agent_id)
        if success:
            return {
                "success": True,
                "message": "Agent unregistered successfully",
                "extensions_cleaned": extensions_removed,
            }
        else:
            raise HTTPException(status_code=404, detail="Agent not found")

    @app.post("/agents/search", response_model=dict[str, Any])
    async def search_agents(request: AgentSearchRequest) -> dict[str, Any]:
        """Search for agents."""
        agents = await storage.search_agents(request.query)
        return {
            "agents": [dict(agent) for agent in agents],
            "count": len(agents),
            "query": request.query,
        }

    # Extension discovery endpoints
    @app.get("/extensions", response_model=dict[str, Any])
    async def list_extensions(
        uri_pattern: str | None = Query(
            None, description="Filter extensions by URI pattern"
        ),
        declaring_agents: list[str] | None = Query(
            None, description="Filter by declaring agents"
        ),
        trust_levels: list[str] | None = Query(
            None, description="Filter by trust levels"
        ),
        page_size: int = Query(
            100, description="Number of extensions per page", ge=1, le=1000
        ),
        page_token: str | None = Query(None, description="Page token for pagination"),
    ) -> dict[str, Any]:
        """List all extensions with provenance information."""
        try:
            extensions, next_page_token, total_count = await storage.list_extensions(
                uri_pattern=uri_pattern,
                declaring_agents=declaring_agents,
                trust_levels=trust_levels,
                page_size=page_size,
                page_token=page_token,
            )

            return {
                "extensions": [ext.to_dict() for ext in extensions],
                "count": len(extensions),
                "total_count": total_count,
                "next_page_token": next_page_token,
                "dev_mode": config.dev_mode,
            }
        except Exception as e:
            logger.error(f"Error listing extensions: {e}")
            raise HTTPException(status_code=500, detail=str(e)) from e

    @app.get("/extensions/{uri:path}", response_model=dict[str, Any])
    async def get_extension_info(uri: str) -> dict[str, Any]:
        """Get specific extension information by URI."""
        try:
            # URL decode the URI
            decoded_uri = unquote(uri)
            extension_info = await storage.get_extension(decoded_uri)

            if extension_info:
                return {
                    "extension_info": extension_info.to_dict(),
                    "found": True,
                }
            else:
                raise HTTPException(status_code=404, detail="Extension not found")

        except HTTPException:
            raise
        except Exception as e:
            logger.error(f"Error getting extension info for {uri}: {e}")
            raise HTTPException(status_code=500, detail=str(e)) from e

    @app.get("/agents/{agent_id}/extensions", response_model=dict[str, Any])
    async def get_agent_extensions(agent_id: str) -> dict[str, Any]:
        """Get all extensions used by a specific agent."""
        try:
            # Verify agent exists
            agent_card = await storage.get_agent(agent_id)
            if not agent_card:
                raise HTTPException(status_code=404, detail="Agent not found")

            extensions = await storage.get_agent_extensions(agent_id)

            return {
                "agent_id": agent_id,
                "extensions": [ext.to_dict() for ext in extensions],
                "count": len(extensions),
            }
        except HTTPException:
            raise
        except Exception as e:
            logger.error(f"Error getting extensions for agent {agent_id}: {e}")
            raise HTTPException(status_code=500, detail=str(e)) from e

    @app.get("/health")
    async def health_check() -> dict[str, str]:
        """Health check endpoint."""
        return {"status": "healthy", "service": "A2A Registry"}

    @app.post("/jsonrpc")
    async def jsonrpc_endpoint(request: Request) -> JSONResponse:
        """JSON-RPC endpoint - primary A2A protocol transport."""
        # Import here to avoid circular imports

        # Get request body
        data = await request.body()

        # Dispatch to JSON-RPC handlers
        response = await async_dispatch(data.decode())

        # The response from jsonrpcserver is a string, parse it to return proper JSON
        import json

        response_data = json.loads(response) if isinstance(response, str) else response

        return JSONResponse(content=response_data, media_type="application/json")

    @app.get("/")
    async def root() -> dict[str, Any]:
        """Root endpoint with service information."""
        protocols = {
            "primary": {
                "transport": "JSONRPC",
                "endpoint": "/jsonrpc",
                "description": "JSON-RPC 2.0 endpoint (A2A default)",
            },
            "secondary": {
                "transport": "HTTP+JSON",
                "endpoints": {
                    "register": "POST /agents",
                    "get": "GET /agents/{id}",
                    "list": "GET /agents",
                    "search": "POST /agents/search",
                    "unregister": "DELETE /agents/{id}",
                    "list_extensions": "GET /extensions",
                    "get_extension": "GET /extensions/{uri}",
                    "agent_extensions": "GET /agents/{id}/extensions",
                },
                "description": "REST API endpoints (convenience)",
            },
        }

        if GRAPHQL_AVAILABLE:
            protocols["graphql"] = {
                "transport": "GraphQL",
                "endpoint": "/graphql",
                "playground": "/graphql",
                "websocket": "/graphql/ws",
                "description": "GraphQL API for AgentExtension system with advanced querying, subscriptions, and analytics",
            }

        return {
            "service": "A2A Registry",
            "version": __version__,
            "description": "Agent-to-Agent Registry Service with dual transport support"
            + (" + GraphQL" if GRAPHQL_AVAILABLE else ""),
            "protocols": protocols,
            "mode": {
                "development": config.dev_mode,
                "extension_verification": config.require_extension_verification,
                "domain_verification": config.require_domain_verification,
                "signature_verification": config.require_signature_verification,
            },
            "health_check": "/health",
            "documentation": "/docs",
        }

    return app

Storage Module

Storage module for A2A Registry.

logger = logging.getLogger(__name__) module-attribute

storage = get_vector_enhanced_storage() module-attribute

ExtensionInfo(uri, description='', required=False, params=None, first_declared_by_agent='', first_declared_at=None, trust_level='TRUST_LEVEL_UNVERIFIED')

Information about an agent extension with provenance tracking.

Source code in src/a2a_registry/storage.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    uri: str,
    description: str = "",
    required: bool = False,
    params: dict | None = None,
    first_declared_by_agent: str = "",
    first_declared_at: datetime | None = None,
    trust_level: str = "TRUST_LEVEL_UNVERIFIED",
):
    self.uri = uri
    self.description = description
    self.required = required
    self.params = params or {}
    self.first_declared_by_agent = first_declared_by_agent
    self.first_declared_at = first_declared_at or datetime.now(UTC)
    self.trust_level = trust_level
    self.declaring_agents: set[str] = set()
    if first_declared_by_agent:
        self.declaring_agents.add(first_declared_by_agent)

declaring_agents = set() instance-attribute

description = description instance-attribute

first_declared_at = first_declared_at or datetime.now(UTC) instance-attribute

first_declared_by_agent = first_declared_by_agent instance-attribute

params = params or {} instance-attribute

required = required instance-attribute

trust_level = trust_level instance-attribute

uri = uri instance-attribute

usage_count property

Number of agents using this extension.

add_declaring_agent(agent_id)

Add an agent to the list of agents using this extension.

Source code in src/a2a_registry/storage.py 
45
46
47
def add_declaring_agent(self, agent_id: str) -> None:
    """Add an agent to the list of agents using this extension."""
    self.declaring_agents.add(agent_id)

from_dict(data) classmethod

Create from dictionary representation.

Source code in src/a2a_registry/storage.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@classmethod
def from_dict(cls, data: dict) -> "ExtensionInfo":
    """Create from dictionary representation."""
    ext_info = cls(
        uri=data["uri"],
        description=data.get("description", ""),
        required=data.get("required", False),
        params=data.get("params", {}),
        first_declared_by_agent=data.get("first_declared_by_agent", ""),
        first_declared_at=datetime.fromisoformat(data["first_declared_at"]),
        trust_level=data.get("trust_level", "TRUST_LEVEL_UNVERIFIED"),
    )
    ext_info.declaring_agents = set(data.get("declaring_agents", []))
    return ext_info

remove_declaring_agent(agent_id)

Remove an agent from the list of agents using this extension.

Source code in src/a2a_registry/storage.py 
49
50
51
def remove_declaring_agent(self, agent_id: str) -> None:
    """Remove an agent from the list of agents using this extension."""
    self.declaring_agents.discard(agent_id)

to_dict()

Convert to dictionary representation.

Source code in src/a2a_registry/storage.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
def to_dict(self) -> dict:
    """Convert to dictionary representation."""
    return {
        "uri": self.uri,
        "description": self.description,
        "required": self.required,
        "params": self.params,
        "first_declared_by_agent": self.first_declared_by_agent,
        "first_declared_at": self.first_declared_at.isoformat(),
        "trust_level": self.trust_level,
        "declaring_agents": list(self.declaring_agents),
        "usage_count": self.usage_count,
    }

FileStorage(data_dir='/data')

Bases: StorageBackend

File-based persistent storage for agent registry.

Source code in src/a2a_registry/storage.py 
321
322
323
324
325
326
327
328
329
def __init__(self, data_dir: str = "/data") -> None:
    self.data_dir = Path(data_dir)
    self.data_dir.mkdir(parents=True, exist_ok=True)
    self.agents_file = self.data_dir / "agents.json"
    self.extensions_file = self.data_dir / "extensions.json"
    self._agents: dict[str, AgentCard] = {}
    self._extensions: dict[str, ExtensionInfo] = {}
    self._load_agents()
    self._load_extensions()

agents_file = self.data_dir / 'agents.json' instance-attribute

data_dir = Path(data_dir) instance-attribute

extensions_file = self.data_dir / 'extensions.json' instance-attribute

get_agent(agent_id) async

Get an agent by ID.

Source code in src/a2a_registry/storage.py 
393
394
395
async def get_agent(self, agent_id: str) -> AgentCard | None:
    """Get an agent by ID."""
    return self._agents.get(agent_id)

get_agent_extensions(agent_id) async

Get all extensions used by a specific agent.

Source code in src/a2a_registry/storage.py 
486
487
488
489
490
async def get_agent_extensions(self, agent_id: str) -> list[ExtensionInfo]:
    """Get all extensions used by a specific agent."""
    return [
        ext for ext in self._extensions.values() if agent_id in ext.declaring_agents
    ]

get_extension(uri) async

Get extension information by URI.

Source code in src/a2a_registry/storage.py 
437
438
439
async def get_extension(self, uri: str) -> ExtensionInfo | None:
    """Get extension information by URI."""
    return self._extensions.get(uri)

list_agents() async

List all registered agents.

Source code in src/a2a_registry/storage.py 
397
398
399
async def list_agents(self) -> list[AgentCard]:
    """List all registered agents."""
    return list(self._agents.values())

list_extensions(uri_pattern=None, declaring_agents=None, trust_levels=None, page_size=100, page_token=None) async

List extensions with optional filtering and pagination.

Source code in src/a2a_registry/storage.py 
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
async def list_extensions(
    self,
    uri_pattern: str | None = None,
    declaring_agents: list[str] | None = None,
    trust_levels: list[str] | None = None,
    page_size: int = 100,
    page_token: str | None = None,
) -> tuple[list[ExtensionInfo], str | None, int]:
    """List extensions with optional filtering and pagination."""
    extensions = list(self._extensions.values())

    # Apply filters
    if uri_pattern:
        extensions = [
            ext for ext in extensions if uri_pattern.lower() in ext.uri.lower()
        ]

    if declaring_agents:
        extensions = [
            ext
            for ext in extensions
            if any(agent in ext.declaring_agents for agent in declaring_agents)
        ]

    if trust_levels:
        extensions = [ext for ext in extensions if ext.trust_level in trust_levels]

    # Simple pagination
    total_count = len(extensions)
    start_idx = 0
    if page_token:
        try:
            start_idx = int(page_token)
        except ValueError:
            start_idx = 0

    end_idx = start_idx + page_size
    page_extensions = extensions[start_idx:end_idx]

    next_page_token = None
    if end_idx < total_count:
        next_page_token = str(end_idx)

    return page_extensions, next_page_token, total_count

register_agent(agent_card) async

Register an agent in the registry.

Source code in src/a2a_registry/storage.py 
383
384
385
386
387
388
389
390
391
async def register_agent(self, agent_card: AgentCard) -> bool:
    """Register an agent in the registry."""
    agent_id = agent_card.get("name")
    if not agent_id:
        return False
    self._agents[agent_id] = agent_card
    self._save_agents()
    logger.info(f"Registered agent: {agent_id}")
    return True

remove_agent_from_extensions(agent_id) async

Remove agent from all extension declarations.

Source code in src/a2a_registry/storage.py 
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
async def remove_agent_from_extensions(self, agent_id: str) -> bool:
    """Remove agent from all extension declarations."""
    extensions_to_remove = []
    modified = False

    for uri, ext_info in self._extensions.items():
        if agent_id in ext_info.declaring_agents:
            ext_info.remove_declaring_agent(agent_id)
            modified = True
            # If no agents are using this extension anymore, remove it
            if ext_info.usage_count == 0:
                extensions_to_remove.append(uri)

    # Remove unused extensions
    for uri in extensions_to_remove:
        del self._extensions[uri]
        logger.info(f"Removed unused extension: {uri}")
        modified = True

    if modified:
        self._save_extensions()

    return True

search_agents(query) async

Search agents by name, description, or capabilities.

Source code in src/a2a_registry/storage.py 
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
async def search_agents(self, query: str) -> list[AgentCard]:
    """Search agents by name, description, or capabilities."""
    results = []
    query_lower = query.lower()

    for agent in self._agents.values():
        # Search in name, description, and skills
        if (
            query_lower in agent.get("name", "").lower()
            or query_lower in agent.get("description", "").lower()
            or any(
                query_lower in skill.get("id", "").lower()
                for skill in agent.get("skills", [])
            )
        ):
            results.append(agent)

    return results

store_extension(extension_info) async

Store extension information.

Source code in src/a2a_registry/storage.py 
430
431
432
433
434
435
async def store_extension(self, extension_info: ExtensionInfo) -> bool:
    """Store extension information."""
    self._extensions[extension_info.uri] = extension_info
    self._save_extensions()
    logger.info(f"Stored extension: {extension_info.uri}")
    return True

unregister_agent(agent_id) async

Unregister an agent.

Source code in src/a2a_registry/storage.py 
401
402
403
404
405
406
407
408
async def unregister_agent(self, agent_id: str) -> bool:
    """Unregister an agent."""
    if agent_id in self._agents:
        del self._agents[agent_id]
        self._save_agents()
        logger.info(f"Unregistered agent: {agent_id}")
        return True
    return False

update_agent_extensions(agent_id, extensions) async

Update extensions for an agent.

Source code in src/a2a_registry/storage.py 
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
async def update_agent_extensions(
    self, agent_id: str, extensions: list[dict]
) -> bool:
    """Update extensions for an agent."""
    # Remove agent from all current extensions
    await self.remove_agent_from_extensions(agent_id)

    # Add agent to new extensions
    for ext_data in extensions:
        uri = ext_data.get("uri", "")
        if not uri:
            continue

        # Check if extension is allowed in current mode
        if not config.is_extension_allowed(uri):
            logger.warning(f"Extension {uri} not allowed in current mode")
            continue

        existing_ext = await self.get_extension(uri)
        if existing_ext:
            existing_ext.add_declaring_agent(agent_id)
            self._save_extensions()  # Save after modification
        else:
            # Create new extension info
            ext_info = ExtensionInfo(
                uri=uri,
                description=ext_data.get("description", ""),
                required=ext_data.get("required", False),
                params=ext_data.get("params", {}),
                first_declared_by_agent=agent_id,
                trust_level=config.get_default_trust_level(),
            )
            await self.store_extension(ext_info)

    return True

InMemoryStorage()

Bases: StorageBackend

In-memory storage for agent registry.

Source code in src/a2a_registry/storage.py 
155
156
157
def __init__(self) -> None:
    self._agents: dict[str, AgentCard] = {}
    self._extensions: dict[str, ExtensionInfo] = {}

get_agent(agent_id) async

Get an agent by ID.

Source code in src/a2a_registry/storage.py 
168
169
170
async def get_agent(self, agent_id: str) -> AgentCard | None:
    """Get an agent by ID."""
    return self._agents.get(agent_id)

get_agent_extensions(agent_id) async

Get all extensions used by a specific agent.

Source code in src/a2a_registry/storage.py 
259
260
261
262
263
async def get_agent_extensions(self, agent_id: str) -> list[ExtensionInfo]:
    """Get all extensions used by a specific agent."""
    return [
        ext for ext in self._extensions.values() if agent_id in ext.declaring_agents
    ]

get_extension(uri) async

Get extension information by URI.

Source code in src/a2a_registry/storage.py 
210
211
212
async def get_extension(self, uri: str) -> ExtensionInfo | None:
    """Get extension information by URI."""
    return self._extensions.get(uri)

list_agents() async

List all registered agents.

Source code in src/a2a_registry/storage.py 
172
173
174
async def list_agents(self) -> list[AgentCard]:
    """List all registered agents."""
    return list(self._agents.values())

list_extensions(uri_pattern=None, declaring_agents=None, trust_levels=None, page_size=100, page_token=None) async

List extensions with optional filtering and pagination.

Source code in src/a2a_registry/storage.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
async def list_extensions(
    self,
    uri_pattern: str | None = None,
    declaring_agents: list[str] | None = None,
    trust_levels: list[str] | None = None,
    page_size: int = 100,
    page_token: str | None = None,
) -> tuple[list[ExtensionInfo], str | None, int]:
    """List extensions with optional filtering and pagination."""
    extensions = list(self._extensions.values())

    # Apply filters
    if uri_pattern:
        extensions = [
            ext for ext in extensions if uri_pattern.lower() in ext.uri.lower()
        ]

    if declaring_agents:
        extensions = [
            ext
            for ext in extensions
            if any(agent in ext.declaring_agents for agent in declaring_agents)
        ]

    if trust_levels:
        extensions = [ext for ext in extensions if ext.trust_level in trust_levels]

    # Simple pagination (in production, use more sophisticated approach)
    total_count = len(extensions)
    start_idx = 0
    if page_token:
        try:
            start_idx = int(page_token)
        except ValueError:
            start_idx = 0

    end_idx = start_idx + page_size
    page_extensions = extensions[start_idx:end_idx]

    next_page_token = None
    if end_idx < total_count:
        next_page_token = str(end_idx)

    return page_extensions, next_page_token, total_count

register_agent(agent_card) async

Register an agent in the registry.

Source code in src/a2a_registry/storage.py 
159
160
161
162
163
164
165
166
async def register_agent(self, agent_card: AgentCard) -> bool:
    """Register an agent in the registry."""
    agent_id = agent_card.get("name")
    if not agent_id:
        return False
    self._agents[agent_id] = agent_card
    logger.info(f"Registered agent: {agent_id}")
    return True

remove_agent_from_extensions(agent_id) async

Remove agent from all extension declarations.

Source code in src/a2a_registry/storage.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
async def remove_agent_from_extensions(self, agent_id: str) -> bool:
    """Remove agent from all extension declarations."""
    extensions_to_remove = []

    for uri, ext_info in self._extensions.items():
        ext_info.remove_declaring_agent(agent_id)
        # If no agents are using this extension anymore, remove it
        if ext_info.usage_count == 0:
            extensions_to_remove.append(uri)

    # Remove unused extensions
    for uri in extensions_to_remove:
        del self._extensions[uri]
        logger.info(f"Removed unused extension: {uri}")

    return True

search_agents(query) async

Search agents by name, description, or capabilities.

Source code in src/a2a_registry/storage.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
async def search_agents(self, query: str) -> list[AgentCard]:
    """Search agents by name, description, or capabilities."""
    results = []
    query_lower = query.lower()

    for agent in self._agents.values():
        # Search in name, description, and skills
        if (
            query_lower in agent.get("name", "").lower()
            or query_lower in agent.get("description", "").lower()
            or any(
                query_lower in skill.get("id", "").lower()
                for skill in agent.get("skills", [])
            )
        ):
            results.append(agent)

    return results

store_extension(extension_info) async

Store extension information.

Source code in src/a2a_registry/storage.py 
204
205
206
207
208
async def store_extension(self, extension_info: ExtensionInfo) -> bool:
    """Store extension information."""
    self._extensions[extension_info.uri] = extension_info
    logger.info(f"Stored extension: {extension_info.uri}")
    return True

unregister_agent(agent_id) async

Unregister an agent.

Source code in src/a2a_registry/storage.py 
176
177
178
179
180
181
182
async def unregister_agent(self, agent_id: str) -> bool:
    """Unregister an agent."""
    if agent_id in self._agents:
        del self._agents[agent_id]
        logger.info(f"Unregistered agent: {agent_id}")
        return True
    return False

update_agent_extensions(agent_id, extensions) async

Update extensions for an agent.

Source code in src/a2a_registry/storage.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
async def update_agent_extensions(
    self, agent_id: str, extensions: list[dict]
) -> bool:
    """Update extensions for an agent."""
    # Remove agent from all current extensions
    await self.remove_agent_from_extensions(agent_id)

    # Add agent to new extensions
    for ext_data in extensions:
        uri = ext_data.get("uri", "")
        if not uri:
            continue

        # Check if extension is allowed in current mode
        if not config.is_extension_allowed(uri):
            logger.warning(f"Extension {uri} not allowed in current mode")
            continue

        existing_ext = await self.get_extension(uri)
        if existing_ext:
            existing_ext.add_declaring_agent(agent_id)
        else:
            # Create new extension info
            ext_info = ExtensionInfo(
                uri=uri,
                description=ext_data.get("description", ""),
                required=ext_data.get("required", False),
                params=ext_data.get("params", {}),
                first_declared_by_agent=agent_id,
                trust_level=config.get_default_trust_level(),
            )
            await self.store_extension(ext_info)

    return True

StorageBackend

Bases: ABC

Abstract base class for storage backends.

get_agent(agent_id) abstractmethod async

Get an agent by ID.

Source code in src/a2a_registry/storage.py 
91
92
93
94
@abstractmethod
async def get_agent(self, agent_id: str) -> AgentCard | None:
    """Get an agent by ID."""
    pass

get_agent_extensions(agent_id) abstractmethod async

Get all extensions used by a specific agent.

Source code in src/a2a_registry/storage.py 
134
135
136
137
@abstractmethod
async def get_agent_extensions(self, agent_id: str) -> list[ExtensionInfo]:
    """Get all extensions used by a specific agent."""
    pass

get_extension(uri) abstractmethod async

Get extension information by URI.

Source code in src/a2a_registry/storage.py 
117
118
119
120
@abstractmethod
async def get_extension(self, uri: str) -> ExtensionInfo | None:
    """Get extension information by URI."""
    pass

list_agents() abstractmethod async

List all registered agents.

Source code in src/a2a_registry/storage.py 
96
97
98
99
@abstractmethod
async def list_agents(self) -> list[AgentCard]:
    """List all registered agents."""
    pass

list_extensions(uri_pattern=None, declaring_agents=None, trust_levels=None, page_size=100, page_token=None) abstractmethod async

List extensions with optional filtering and pagination.

Source code in src/a2a_registry/storage.py 
122
123
124
125
126
127
128
129
130
131
132
@abstractmethod
async def list_extensions(
    self,
    uri_pattern: str | None = None,
    declaring_agents: list[str] | None = None,
    trust_levels: list[str] | None = None,
    page_size: int = 100,
    page_token: str | None = None,
) -> tuple[list[ExtensionInfo], str | None, int]:
    """List extensions with optional filtering and pagination."""
    pass

register_agent(agent_card) abstractmethod async

Register an agent in the registry.

Source code in src/a2a_registry/storage.py 
86
87
88
89
@abstractmethod
async def register_agent(self, agent_card: AgentCard) -> bool:
    """Register an agent in the registry."""
    pass

remove_agent_from_extensions(agent_id) abstractmethod async

Remove agent from all extension declarations.

Source code in src/a2a_registry/storage.py 
146
147
148
149
@abstractmethod
async def remove_agent_from_extensions(self, agent_id: str) -> bool:
    """Remove agent from all extension declarations."""
    pass

search_agents(query) abstractmethod async

Search agents by name, description, or capabilities.

Source code in src/a2a_registry/storage.py 
106
107
108
109
@abstractmethod
async def search_agents(self, query: str) -> list[AgentCard]:
    """Search agents by name, description, or capabilities."""
    pass

store_extension(extension_info) abstractmethod async

Store extension information.

Source code in src/a2a_registry/storage.py 
112
113
114
115
@abstractmethod
async def store_extension(self, extension_info: ExtensionInfo) -> bool:
    """Store extension information."""
    pass

unregister_agent(agent_id) abstractmethod async

Unregister an agent.

Source code in src/a2a_registry/storage.py 
101
102
103
104
@abstractmethod
async def unregister_agent(self, agent_id: str) -> bool:
    """Unregister an agent."""
    pass

update_agent_extensions(agent_id, extensions) abstractmethod async

Update extensions for an agent.

Source code in src/a2a_registry/storage.py 
139
140
141
142
143
144
@abstractmethod
async def update_agent_extensions(
    self, agent_id: str, extensions: list[dict]
) -> bool:
    """Update extensions for an agent."""
    pass

get_storage_backend()

Get the appropriate storage backend based on environment configuration.

Source code in src/a2a_registry/storage.py 
553
554
555
556
557
558
559
560
561
562
563
def get_storage_backend() -> StorageBackend:
    """Get the appropriate storage backend based on environment configuration."""
    storage_type = config.storage_type
    data_dir = config.storage_data_dir

    if storage_type == "file":
        logger.info(f"Using file storage backend with data directory: {data_dir}")
        return FileStorage(data_dir)
    else:
        logger.info("Using in-memory storage backend")
        return InMemoryStorage()

get_vector_enhanced_storage()

Get vector-enhanced storage wrapper.

Source code in src/a2a_registry/storage.py 
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
def get_vector_enhanced_storage() -> StorageBackend:
    """Get vector-enhanced storage wrapper."""
    try:
        from .vector_enhanced_storage import VectorEnhancedStorage

        backend = get_storage_backend()
        vector_storage = VectorEnhancedStorage(backend)
        logger.info("Using vector-enhanced storage with FAISS")
        return vector_storage
    except ImportError as e:
        logger.warning(f"Vector search dependencies not available: {e}")
        logger.info("Falling back to basic storage backend")
        return get_storage_backend()
    except Exception as e:
        logger.error(f"Failed to initialize vector-enhanced storage: {e}")
        logger.info("Falling back to basic storage backend")
        return get_storage_backend()

CLI Module

Command line interface for a2a-registry.

cli()

A2A Registry CLI.

Source code in src/a2a_registry/cli.py 
 7
 8
 9
10
@click.group()
def cli() -> None:
    """A2A Registry CLI."""
    pass

main()

Main entry point for the CLI.

Source code in src/a2a_registry/cli.py 
29
30
31
def main() -> None:
    """Main entry point for the CLI."""
    cli()

serve(host, port, reload)

Start the A2A Registry JSON-RPC server.

Source code in src/a2a_registry/cli.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@cli.command()
@click.option("--host", default="127.0.0.1", help="Host to bind to")
@click.option("--port", default=8000, help="Port to bind to")
@click.option("--reload", is_flag=True, help="Enable auto-reload")
def serve(host: str, port: int, reload: bool) -> None:
    """Start the A2A Registry JSON-RPC server."""
    click.echo(f"Starting A2A Registry server on {host}:{port}")
    uvicorn.run(
        "a2a_registry.server:create_app",
        host=host,
        port=port,
        reload=reload,
        factory=True,
    )

Extension System

URI Allowlist

  • Configurable whitelist for agent extension URIs
  • Trust level assignment for extensions
  • Granular access control

Extension Registration

# Example extension registration
extension = {
    "uri": "https://example.com/agent-extension",
    "trust_level": "high",
    "allowed_skills": ["weather", "forecast"]
}
registry.register_extension(extension)

Error Handling

Common Error Codes

  • AGENT_REGISTRATION_FAILED (1001)
  • AGENT_NOT_FOUND (1002)
  • UNAUTHORIZED_ACCESS (1003)
  • INVALID_PROTOCOL_VERSION (1004)

Error Response Example

{
  "error": {
    "code": 1001,
    "message": "Agent registration failed",
    "details": {
      "reason": "Invalid agent card format"
    }
  }
}

Performance & Rate Limiting

  • Default rate limit: 100 requests/minute
  • Configurable per API key
  • Supports exponential backoff

Monitoring & Health Checks

Agent Health Endpoints

  • /health: Overall system health
  • /agents/health: Aggregate agent health
  • /metrics: Prometheus-compatible metrics

Configuration Options

# Example configuration
registry_config = {
    "host": "0.0.0.0",
    "port": 8000,
    "log_level": "INFO",
    "auth_mode": "production",
    "trust_levels": {
        "development": {"max_agents": 10},
        "production": {"max_agents": 1000}
    }
}

Compatibility

  • A2A Protocol: v0.3.0
  • Python: 3.9+
  • Supported Transports:
  • JSON-RPC 2.0 (Primary)
  • REST
  • GraphQL
  • gRPC (Experimental)

Best Practices

  1. Use JWT for authentication in production
  2. Implement agent health checks
  3. Validate agent cards before registration
  4. Use GraphQL for complex querying
  5. Implement proper error handling

SDK & Client Libraries

  • Official Python SDK
  • Community-supported libraries for other languages

Troubleshooting

Common Issues

  • Incorrect agent card format
  • Authentication failures
  • Protocol version mismatches

Refer to our Troubleshooting Guide for detailed resolution steps.