In bedrock agentcore, we have harness and runtime. We going to look at the code differences trying to invoke them.
Whenever possible, ry to use 'bedrock-agentcore' as it is the newer package atleast for now.
Older library typically uses bedrock-agent-runtime.
https://docs.aws.amazon.com/boto3/latest/reference/services/bedrock-agent-runtime.html
Harness
Notice that everything is the same except we need to update harnessArn and we invoke the method invoke_harness(). Notice we are instantiating 'bedrick-agentcore'. There's also a client called 'bedrock-agent-runtime.
And when you see the word harness - you know you're in a good seat. :)
import boto3
import json
import uuid
client = boto3.client('bedrock-agentcore', region_name='ap-southeast-2')
session_id = str(uuid.uuid4())
response = client.invoke_harness(
harnessArn='your-harness-arn',
runtimeSessionId=session_id,
messages=[
{
'role': 'user',
'content': [{'text': 'Hello, how can you help me?'}]
}
]
)
# Process the streaming response
for event in response['stream']:
if 'contentBlockDelta' in event:
delta = event['contentBlockDelta'].get('delta', {})
if 'text' in delta:
print(delta['text'], end='')
print()
By default, every session includes the shell tool (for executing bash commands) and the file_operations tool (for viewing, creating, and editing files). You can restrict or disable these default tools at any time using the allowedTools parameter.
To learn how to pass tools, please refer to the doc here
https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/harness-tools.html
Runtime
Here we are calling invoke_agent_runtime() and pass in the relevant parameters agentRuntimeArn, runtimeSessionId and Endpoint.
The purpsoe of endpoint is to expose different version of your agent. This acts like a deployement slot for your app service - avoiding interruption to existing runtime that has been deployed.
import boto3
import json
import uuid
client = boto3.client('bedrock-agentcore', region_name='ap-southeast-2')
payload = json.dumps({"prompt": "Explain machine learning in simple terms"})
session_id = str(uuid.uuid4())
response = client.invoke_agent_runtime(
agentRuntimeArn='your-agent-runtime-arn', # Replace with your Agent Runtime ARN
runtimeSessionId=session_id, # Must be 33+ char. Every new SessionId will create a new MicroVM
payload=payload,
qualifier="<Replace with your Endpoint>" # This is Optional. When the field is not provided, Runtime will use DEFAULT endpoint
)
response_body = response['response'].read()
response_data = json.loads(response_body)
print("Agent Response:", response_data)
Invoking knowledgebased
You can use the following code to your agent to use knowledge-based.
# agent runtime with knowledge base retrieval
import boto3
client = boto3.client("bedrock-agent-runtime", region_name="ap-southeast-2")
# AgenticRetrieveStream - streaming, agent-driven multi-step retrieval
response = client.agentic_retrieve_stream(
messages=[{"role": "user", "content": {"text": "your query here"}}],
retrievers=[{"configuration": {"knowledgeBase": {"knowledgeBaseId": "YOUR-KNOWLEDGE-BASE-ID"}}}],
agenticRetrieveConfiguration={"foundationModelType": "MANAGED", "maxAgentIteration": 5},
generateResponse=True,
)
for event in response["stream"]:
if "responseEvent" in event: # generated answer, streamed token-by-token
print(event["responseEvent"].get("text", ""), end="", flush=True)
elif "result" in event: # final retrieved chunks and citations
for item in event["result"].get("results", []):
print("Source:", item.get("content", {}).get("text"))
elif "traceEvent" in event: # agent planning / retrieval steps
attrs = event["traceEvent"].get("attributes", {})
print(f"[{attrs.get('step')}/{attrs.get('status')}] {attrs.get('message')}")
retrieve = client.retrieve(
knowledgeBaseId="YOUR-KNOWLEDGE-BASE-ID",
retrievalQuery={"text": "your query here"},
)
print("Retrieve:", retrieve["retrievalResults"])
It is important to checkout different packages available like bedrock, bedrock-runtime here
https://docs.aws.amazon.com/bedrock-agentcore/
Comments