I’m running a CrewAI agent that chains primarily MCP tools, some of them return JWTs from an MCP server. Those JWTs need to be passed exactly to the next tool. The problem is, the agent will mutate a JWT right before it uses it as input to the next tool call. This happens 5% of the time. Sometimes the JWT is truncated, sometimes 100 characters are appended, sometimes it’s just 1 character that is different.
async def run_agent(user_prompt: str):
static_servers = [
{"url": os.getenv("SKYFIRE_MCP_URL"), "transport": "sse", "headers": {"skyfire-api-key": os.getenv("SKYFIRE_API_KEY")}},
{"url": os.getenv("VISUALISATION_MCP_URL"), "transport": "sse", "headers": {}}
]
dynamic_servers = []
agent_memory = {"steps": []}
found_servers = True
while found_servers:
all_servers = static_servers + dynamic_servers
try:
resources = [get_all_resources(server.get("url"), server.get("headers", {})) for server in all_servers]
all_resources = await asyncio.gather(*resources)
agent_memory['resource_guide'] = "\n".join(resource for resource in all_resources if resource)
resource_knowledge = StringKnowledgeSource(content=agent_memory['resource_guide'])
with MCPServerAdapter(all_servers) as mcp_tools:
connect_mcp_server_tool = ConnectToMCPTool(result_as_answer=True)
all_tools = mcp_tools + [connect_mcp_server_tool]
# Agent initialization
agent = Agent(
role="Autonomous E-commerce Buyer Agent",
goal=user_prompt,
backstory=dedent(f"""An expert agent that navigates complex purchasing workflows using MCP tools, who never mutates, appends to, or truncates any tokens."""),
tools=all_tools,
verbose=True,
knowledge_sources=[resource_knowledge],
allow_delegation=False,
max_iter=10
)
# Task initialization
task = Task(
description=dedent(f"""
Memory: {agent_memory.get('steps', '')}
Avoid repeating tool calls that have already been executed.
Use newest obtained tool outputs as inputs to the next tool calls.
When using a seller's tools, ensure that you have created an account and logged in if necessary.
Ensure that you never mutate, append, or truncate any tokens.
Never assume dataset URLs.
When there is a need to create an account, use 123456S$d#d as the password.
Generate a random UUID buyer tag for Skyfire token calls.
"""),
expected_output="The final result or confirmation of the requested action.",
agent=agent
)
def step_callback(step):
record = {
"type": type(step).__name__,
"data": step.__dict__,
}
agent_memory["steps"].append(record)
pprint(record)
crew = Crew(
agents=[agent],
tasks=[task],
process=Process.sequential,
verbose=False,
step_callback=step_callback # Function is called after each agent step
)
result = await crew.kickoff_async()
try:
parsed_result = json.loads(result.raw)
if parsed_result.get("action") == "connect_to_mcp":
mcp_url = parsed_result.get("server_url")
if mcp_url and not any(s['url'] == mcp_url for s in dynamic_servers):
print(f"🆕 Adding new server to dynamic list: {mcp_url}")
dynamic_servers.append({"url": mcp_url, "transport": "sse", "headers": {}})
continue
except (json.JSONDecodeError, AttributeError):
pass
found_servers = False
except Exception as e:
print(f"⚠️ Agent execution failed: {e}")
import traceback
traceback.print_exc()
return f"Agent run failed: {e}"
# pprint(agent_memory)
return result
For example, agent calls MCP tool create-pay-token and gets back:
'Transaction of 0.002 is successful to '
'012683a0-ee16-43f3-b27d-6dfa37976503 via token '
'eyJhbGciO'
Next, within the same task, the agent calls MCP tool charge-token and passes in as tool_input: pay_token: eyJhbGciOanzYt, where the appended characters are random and no where else found in the call flow.