a2aproject / a2aproject/a2a-samples

Push Notification is not receiving the processed output

Abierto
#335 0 comentarios 0 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Jupyter Notebook
Estrellas
1.8k
Forks
751
Métricas de merge de PR
Sin PR fusionados en 30 d

Descripción

### What happened?

I am using the langgraph sample code from the GitHub samples folder. I wrote a non-blocking client which uses push notification. PFB the client code

`
async def main() -> None:
# Configure logging to show INFO level messages
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__) # Get a logger instance

# --8<-- [start:A2ACardResolver]

base_url = 'http://localhost:10000'

async with httpx.AsyncClient() as httpx_client:
# Initialize A2ACardResolver
resolver = A2ACardResolver(
httpx_client=httpx_client,
base_url=base_url,
# agent_card_path uses default, extended_agent_card_path also uses default
)
# --8<-- [end:A2ACardResolver]

# Fetch Public Agent Card and Initialize Client
final_agent_card_to_use: AgentCard | None = None

try:
logger.info(
f'Attempting to fetch public agent card from: {base_url}{AGENT_CARD_WELL_KNOWN_PATH}'
)
_public_card = (
await resolver.get_agent_card()
) # Fetches from default public path
logger.info('Successfully fetched public agent card:')
logger.info(
_public_card.model_dump_json(indent=2, exclude_none=True)
)
final_agent_card_to_use = _public_card
logger.info(
'\nUsing PUBLIC agent card for client initialization (default).'
)

if _public_card.supports_authenticated_extended_card:
try:
logger.info(
'\nPublic card supports authenticated extended card. '
'Attempting to fetch from: '
f'{base_url}{EXTENDED_AGENT_CARD_PATH}'
)
auth_headers_dict = {
'Authorization': 'Bearer dummy-token-for-extended-card'
}
_extended_card = await resolver.get_agent_card(
relative_card_path=EXTENDED_AGENT_CARD_PATH,
http_kwargs={'headers': auth_headers_dict},
)
logger.info(
'Successfully fetched authenticated extended agent card:'
)
logger.info(
_extended_card.model_dump_json(
indent=2, exclude_none=True
)
)
final_agent_card_to_use = (
_extended_card # Update to use the extended card
)
logger.info(
'\nUsing AUTHENTICATED EXTENDED agent card for client '
'initialization.'
)
except Exception as e_extended:
logger.warning(
f'Failed to fetch extended agent card: {e_extended}. '
'Will proceed with public card.',
exc_info=True,
)
elif (
_public_card
): # supports_authenticated_extended_card is False or None
logger.info(
'\nPublic card does not indicate support for an extended card. Using public card.'
)

except Exception as e:
logger.error(
f'Critical error fetching public agent card: {e}', exc_info=True
)
raise RuntimeError(
'Failed to fetch the public agent card. Cannot continue.'
) from e

# --8<-- [start:send_message]
client = A2AClient(
httpx_client=httpx_client, agent_card=final_agent_card_to_use
)
logger.info('A2AClient initialized.')

context_id = uuid4().hex
task_id = None

message = Message(
role='user',
parts=[TextPart(text="how much is 15 USD in INR?")],
message_id=str(uuid4()),
task_id=task_id,
context_id=context_id,
)

payload = MessageSendParams(
id=str(uuid4()),
message=message,
configuration=MessageSendConfiguration(
accepted_output_modes=['text'],
push_notification_config=PushNotificationConfig(
url=f'http://localhost:8081/a2a/notify',
),
blocking=False
),
)

request = SendMessageRequest(
id=str(uuid4()), params=payload
)

response = await client.send_message(request)
print(response.model_dump(mode='json', exclude_none=True))

# Get the task from the response
task = response.root.result
task_id = task.id

terminal = {TaskState.completed, TaskState.failed, TaskState.canceled, TaskState.rejected}
while task.status.state not in terminal:
await asyncio.sleep(0.5)
get_resp = await client.get_task(GetTaskRequest(id=str(uuid4()), params=TaskQueryParams(id=task_id)))
if hasattr(get_resp, "root") and hasattr(get_resp.root, "result"):
task = get_resp.root.result
print("Polled state:", task.status.state)
else:
print("Unexpected response while polling")
break

print("Final state:", task.status.state)
print(task)`

Agent is not pushing the result to the push notification endpoint. I am getting the following output from the push notification endpoint.

`--- A2A NOTIFICATION ---
Auth header: None
Payload: {'contextId': '0d4a27651fa84ab0a44583f91ffc2482', 'history': [{'contextId': '0d4a27651fa84ab0a44583f91ffc2482', 'kind': 'message', 'messageId': 'f7ad0ab0-4605-489e-9543-cff3a7a7060a', 'parts': [{'kind': 'text', 'text': 'how much is 15 USD in INR?'}], 'role': 'user', 'taskId': '9e5bf779-4640-440e-af43-70552d9e6c7c'}], 'id': '9e5bf779-4640-440e-af43-70552d9e6c7c', 'kind': 'task', 'status': {'state': 'submitted'}}
------------------------`

Why is the code not pushing the actual output to the push notification endpoint? If I enable blocking = True, I am getting the output in the notification endpoint.

### Relevant log output

```shell

```

### Code of Conduct

- [x] I agree to follow this project's Code of Conduct

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.