1  """
  2  AI Agent Signed Tool Call Example
  3  =================================
  4  
  5  Demonstrates how an AI agent signs tool calls with Auths action envelopes,
  6  and how a server verifies them before execution.
  7  
  8  Requirements:
  9      pip install auths-verifier
 10  
 11  Security Note:
 12      This example passes key material as hex strings for simplicity. Python
 13      strings are immutable and not zeroizable. For production, store keys in
 14      a secure enclave, hardware security module, or secret manager (e.g.,
 15      AWS KMS, HashiCorp Vault, macOS Keychain).
 16  
 17  Usage:
 18      python langchain_tool.py
 19  """
 20  
 21  import json
 22  import sys
 23  
 24  try:
 25      from auths_verifier import sign_action, verify_action_envelope
 26  except ImportError:
 27      print("Install auths-verifier: pip install auths-verifier")
 28      sys.exit(1)
 29  
 30  
 31  def generate_test_keypair():
 32      """Generate an Ed25519 keypair for demonstration.
 33  
 34      In production, use `auths id create` to generate keys stored in the
 35      system keychain, then export the seed for agent use.
 36      """
 37      try:
 38          from cryptography.hazmat.primitives.asymmetric.ed25519 import (
 39              Ed25519PrivateKey,
 40          )
 41  
 42          private_key = Ed25519PrivateKey.generate()
 43          seed_hex = private_key.private_bytes_raw().hex()
 44          pub_hex = private_key.public_key().public_bytes_raw().hex()
 45          return seed_hex, pub_hex
 46      except ImportError:
 47          print("Install cryptography for key generation: pip install cryptography")
 48          print("(or use a pre-generated key from `auths id create`)")
 49          sys.exit(1)
 50  
 51  
 52  # --- Agent Side ---
 53  
 54  
 55  def agent_execute_tool(agent_seed_hex: str, agent_did: str, tool_name: str, args: dict) -> str:
 56      """Agent signs a tool call and returns the signed envelope.
 57  
 58      This would typically be called by a LangChain tool wrapper or similar
 59      agent framework before sending the request to a server.
 60      """
 61      payload = {"tool": tool_name, "args": args}
 62  
 63      envelope_json = sign_action(
 64          private_key_hex=agent_seed_hex,
 65          action_type="tool_call",
 66          payload_json=json.dumps(payload),
 67          identity_did=agent_did,
 68      )
 69  
 70      return envelope_json
 71  
 72  
 73  # --- Server Side ---
 74  
 75  
 76  def server_verify_and_execute(envelope_json: str, trusted_keys: dict) -> dict:
 77      """Server verifies the signed envelope and executes the tool call.
 78  
 79      Args:
 80          envelope_json: The signed action envelope from the agent
 81          trusted_keys: Mapping of DID -> public key hex for authorized agents
 82  
 83      Returns:
 84          Execution result or error
 85      """
 86      envelope = json.loads(envelope_json)
 87  
 88      # 1. Look up the agent's public key by DID
 89      identity = envelope.get("identity", "")
 90      pub_key_hex = trusted_keys.get(identity)
 91      if pub_key_hex is None:
 92          return {"error": f"Unknown identity: {identity}"}
 93  
 94      # 2. Verify the signature
 95      result = verify_action_envelope(envelope_json, pub_key_hex)
 96      if not result.valid:
 97          return {"error": f"Signature verification failed: {result.error}"}
 98  
 99      # 3. Execute the tool call (application logic)
100      payload = envelope.get("payload", {})
101      tool = payload.get("tool", "unknown")
102      args = payload.get("args", {})
103  
104      print(f"  Verified agent: {identity}")
105      print(f"  Executing tool: {tool}")
106      print(f"  Arguments: {json.dumps(args, indent=2)}")
107  
108      return {"status": "executed", "tool": tool, "result": "success"}
109  
110  
111  # --- Demo ---
112  
113  
114  def main():
115      print("=== Auths AI Agent Signed Tool Call Demo ===\n")
116  
117      # Generate keys for the demo
118      agent_seed, agent_pubkey = generate_test_keypair()
119      agent_did = "did:keri:EAgent123"
120  
121      print(f"Agent DID:        {agent_did}")
122      print(f"Agent public key: {agent_pubkey[:16]}...")
123      print()
124  
125      # Server's trusted key registry
126      trusted_keys = {agent_did: agent_pubkey}
127  
128      # Agent signs a tool call
129      print("1. Agent signs a tool call:")
130      envelope_json = agent_execute_tool(
131          agent_seed_hex=agent_seed,
132          agent_did=agent_did,
133          tool_name="execute_sql",
134          args={"query": "SELECT COUNT(*) FROM users", "database": "analytics"},
135      )
136  
137      envelope = json.loads(envelope_json)
138      print(f"   Type:      {envelope['type']}")
139      print(f"   Identity:  {envelope['identity']}")
140      print(f"   Timestamp: {envelope['timestamp']}")
141      print(f"   Signature: {envelope['signature'][:32]}...")
142      print()
143  
144      # Server verifies and executes
145      print("2. Server verifies and executes:")
146      result = server_verify_and_execute(envelope_json, trusted_keys)
147      print(f"   Result: {result}")
148      print()
149  
150      # Demonstrate rejection of tampered envelope
151      print("3. Tampered envelope is rejected:")
152      tampered = json.loads(envelope_json)
153      tampered["payload"]["args"]["query"] = "DROP TABLE users"
154      tampered_json = json.dumps(tampered)
155  
156      result = server_verify_and_execute(tampered_json, trusted_keys)
157      print(f"   Result: {result}")
158      print()
159  
160      # Demonstrate rejection of unknown identity
161      print("4. Unknown identity is rejected:")
162      unknown = json.loads(envelope_json)
163      unknown["identity"] = "did:keri:EUnknown"
164      unknown_json = json.dumps(unknown)
165  
166      result = server_verify_and_execute(unknown_json, trusted_keys)
167      print(f"   Result: {result}")
168  
169  
170  if __name__ == "__main__":
171      main()