Connect an Agent to DataLinks over A2A

​

Prerequisites

• A DataLinks API token. See Get an API token. The same token you use for the REST API and SDK works for A2A.
• The namespace you want the agent to work against. You can find it under My Data.
• Python 3.10+ (this guide uses the official a2a-sdk client; any A2A-compliant client works).

​

Install Required Dependencies

uv pip install a2a-sdk httpx

uv pip install datalinks

​

Configure Your Environment

• DL_HOST - the DataLinks host that serves the A2A endpoints: https://api.datalinks.com/api/v1
• DL_API_KEY - your DataLinks API token
• DL_NAMESPACE - the namespace the agent should query
• DL_A2A_AGENT - (optional) the agent to use; defaults to the first agent listed by the server

​

How the DataLinks A2A Endpoints Work

​

Connect and Ask a Question

import httpx

async def list_agents(http: httpx.AsyncClient, host: str) -> list[dict]:
    resp = await http.get(f"{host}/a2a")
    resp.raise_for_status()
    return resp.json()

from a2a.client import (
    A2ACardResolver, AuthInterceptor, ClientConfig, ClientFactory,
    ClientCallContext, InMemoryContextCredentialStore,
)

SESSION_ID = "datalinks-a2a-example"

agent_base = f"{host}/a2a/{agent_name}"
card = await A2ACardResolver(http, base_url=agent_base).get_agent_card()

store = InMemoryContextCredentialStore()
for requirement in card.security_requirements:
    for scheme_name in requirement.schemes:
        await store.set_credentials(SESSION_ID, scheme_name, api_key)

factory = ClientFactory(ClientConfig(httpx_client=http, streaming=True))
client = factory.create(card, interceptors=[AuthInterceptor(store)])

from uuid import uuid4
from google.protobuf.struct_pb2 import Struct
from a2a.types import (
    Message, Part, Role, SendMessageConfiguration, SendMessageRequest,
)

metadata = Struct()
metadata.update({"namespace": namespace})   # optionally: "model", "provider"

request = SendMessageRequest(
    message=Message(
        message_id=str(uuid4()),
        role=Role.ROLE_USER,
        parts=[Part(text=question)],
        metadata=metadata,
    ),
    configuration=SendMessageConfiguration(accepted_output_modes=["text", "data"]),
)

context = ClientCallContext(state={"sessionId": SESSION_ID})
async for response in client.send_message(request, context=context):
    render(response)   # see "Handling the streamed response" below

​

Handling the Streamed Response

• nl2ql emits a generated-query artifact (the query, as a TextPart) and a result-data artifact (the rows, as a DataPart).
• assistant streams working status updates while it plans and runs steps, a step-N artifact per step (the step’s query plus its rows), and a final answer artifact (the synthesised answer, as a TextPart).

from google.protobuf.json_format import MessageToDict
from a2a.types import TaskState

def render(response) -> None:
    kind = response.WhichOneof("payload")
    if kind == "message":
        _print_parts(response.message.parts)
    elif kind == "artifact_update":
        _print_parts(response.artifact_update.artifact.parts)
    elif kind == "status_update":
        status = response.status_update.status
        print(f"[status: {TaskState.Name(status.state)}]")
        if status.HasField("message"):
            _print_parts(status.message.parts)
    elif kind == "task":
        task = response.task
        print(f"[task {task.id}: {TaskState.Name(task.status.state)}]")
        for artifact in task.artifacts:
            _print_parts(artifact.parts)

def _print_parts(parts) -> None:
    for part in parts:
        content = part.WhichOneof("content")
        if content == "text":
            print(part.text)
        elif content == "data":
            payload = MessageToDict(part.data)   # protobuf -> native dict/list
            rows = payload.get("rows") if isinstance(payload, dict) else None
            print(rows if rows is not None else payload)

​

Verify the Connection

• GET {host}/a2a returns a non-empty list of agents (nl2ql and assistant).
• Fetching the AgentCard for your chosen agent succeeds.
• A simple question streams back a status_update reaching a completed state, followed by an answer.
• No 401/403 authorization errors appear.

​

Troubleshooting

• Confirm your DataLinks API token is correct and current.
• Make sure the token is sent as Bearer <token> with no extra spaces.
• Confirm the scheme name you registered matches the one the AgentCard declares (bearer).

• Confirm DL_HOST points at https://api.datalinks.com/api/v1 (A2A lives under the API base, not the root host).

• The namespace must live in the message metadata (message.metadata.namespace), not in the URL, the request-level metadata, or the body.

• You’re only handling TextParts. Handle DataParts too (see the warning above); the rows arrive under part.data.rows.

• Increase the HTTP client timeout; some agent runs take longer than the default.
• Confirm the namespace in the message metadata exists, has at least one dataset, and you have access to it.

​

When to Use A2A, MCP, or the SDK

​

Next Steps

• Read the A2A Protocol Reference for the full endpoint, AgentCard, and event details.
• Try the complete runnable client.
• Combine the agent’s answers with your own workflow or multi-agent system.