Last active
October 1, 2025 13:00
-
-
Save tinvaan/8b7b6d1829b96e0bf5896d64578ddd3f to your computer and use it in GitHub Desktop.
Claude Code structured outputs via Hooks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import anyio | |
| import instructor | |
| import json | |
| from claude_agent_sdk import ( | |
| ClaudeAgentOptions, | |
| ClaudeSDKClient, | |
| HookMatcher, | |
| ResultMessage | |
| ) | |
| from claude_agent_sdk.types import HookJSONOutput | |
| from pprint import pprint | |
| from datetime import datetime | |
| from pydantic import BaseModel | |
| class Event(BaseModel): | |
| date: datetime | |
| location: str | |
| description: str | |
| class Query: | |
| async def parse(self, event, *args, **kwargs) -> HookJSONOutput: | |
| def read(file): | |
| transcripts = [] | |
| with open(file, 'r', encoding='utf-8') as f: | |
| for line in f: | |
| msg = json.loads(line) | |
| if msg.get('message', {}).get('role') == 'assistant': | |
| transcripts.extend([ | |
| content.get('text') for content in msg.get('message', {}).get('content') if content.get('type') == 'text' | |
| ]) | |
| return transcripts | |
| assert event.get('transcript_path'), "Transcript file not found" | |
| excerpt = "".join(read(event.get('transcript_path'))[-1]) | |
| client = instructor.from_provider("anthropic/claude-3-5-sonnet-20240620") | |
| self.res = client.chat.completions.create( | |
| response_model=Event, | |
| messages=[ | |
| { | |
| "role": "system", | |
| "content": "Extract the key attributes from the given excerpt" | |
| }, | |
| { | |
| "role": "user", | |
| "content": excerpt | |
| } | |
| ] | |
| ) | |
| return {} | |
| async def extract(self): | |
| opts = ClaudeAgentOptions( | |
| allowed_tools=["Bash"], | |
| hooks={ | |
| "Stop": [ | |
| HookMatcher(matcher="Bash", hooks=[self.parse]) | |
| ] | |
| } | |
| ) | |
| async with ClaudeSDKClient(options=opts) as agent: | |
| await agent.query("What is Tiananmen Square famous for?") | |
| async for msg in agent.receive_response(): | |
| if isinstance(msg, ResultMessage): | |
| msg.result = getattr(self, "res", msg.result) | |
| return msg | |
| async def run(self): | |
| info = await self.extract() | |
| ok = isinstance(info.result, Event) | |
| if ok: | |
| pprint(info.result.model_dump()) | |
| if __name__ == '__main__': | |
| q = Query() | |
| anyio.run(q.run) |
Author
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Examples
○ python test.py {'date': datetime.datetime(1949, 10, 1, 0, 0, tzinfo=TzInfo(UTC)), 'description': "Mao Zedong proclaimed the founding of the People's Republic " 'of China', 'location': 'Tiananmen Square, Beijing, China'}○ python test.py {'date': datetime.datetime(1989, 6, 4, 0, 0, tzinfo=TzInfo(UTC)), 'description': 'Military crackdown on pro-democracy student protests, ' 'resulting in significant casualties. This event became ' 'internationally known and included the iconic "Tank Man" ' 'incident.', 'location': 'Tiananmen Square, Beijing, China'}