a2aproject / a2aproject/a2a-samples

Push Notification is not receiving the processed output

Đang mở
#335 0 bình luận 0 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
Jupyter Notebook
Star
1.8k
Fork
751
Chỉ số merge pull request
Không có pull request nào được merge trong 30 ngày

Mô tả

### 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

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.