Coverage for src / agent_contracts / runtime / context.py: 100%

1"""Request/Response context types for runtime execution.""" 

2from __future__ import annotations 

3 

4from dataclasses import dataclass, field 

5from typing import Any 

6 

7 

8@dataclass 

9class RequestContext: 

10 """Context for an agent execution request. 

11  

12 Contains all information needed to build initial state and execute the graph. 

13  

14 Attributes: 

15 session_id: Unique session identifier 

16 action: The action to perform (e.g., "answer", "propose") 

17 params: Optional parameters for the action 

18 message: Optional user message 

19 image: Optional base64-encoded image 

20 resume_session: Whether to restore previous session state 

21 metadata: Additional app-specific metadata 

22  

23 Example: 

24 >>> ctx = RequestContext( 

25 ... session_id="abc123", 

26 ... action="answer", 

27 ... message="I prefer casual style", 

28 ... resume_session=True, 

29 ... ) 

30 """ 

31 session_id: str 

32 action: str 

33 params: dict[str, Any] | None = None 

34 message: str | None = None 

35 image: str | None = None 

36 resume_session: bool = False 

37 metadata: dict[str, Any] = field(default_factory=dict) 

38 

39 def get_param(self, key: str, default: Any = None) -> Any: 

40 """Get a parameter value safely. 

41  

42 Args: 

43 key: Parameter key 

44 default: Default value if not found 

45  

46 Returns: 

47 Parameter value or default 

48 """ 

49 if self.params is None: 

50 return default 

51 return self.params.get(key, default) 

52 

53 

54@dataclass 

55class ExecutionResult: 

56 """Result of an agent execution. 

57  

58 Contains the final state and response information. 

59  

60 Attributes: 

61 state: The final state after graph execution 

62 response_type: Type of response (e.g., "interview", "proposals") 

63 response_data: Response payload data 

64 response_message: Optional response message 

65 success: Whether execution completed successfully 

66 error: Error message if execution failed 

67  

68 Example: 

69 >>> result = ExecutionResult( 

70 ... state=final_state, 

71 ... response_type="interview", 

72 ... response_data={"question": "What's your style?"}, 

73 ... ) 

74 """ 

75 state: dict[str, Any] 

76 response_type: str | None = None 

77 response_data: dict[str, Any] | None = None 

78 response_message: str | None = None 

79 success: bool = True 

80 error: str | None = None 

81 

82 @classmethod 

83 def from_state(cls, state: dict[str, Any]) -> "ExecutionResult": 

84 """Create ExecutionResult from final state. 

85  

86 Extracts response information from the response slice. 

87  

88 Args: 

89 state: Final state after graph execution 

90  

91 Returns: 

92 ExecutionResult instance 

93 """ 

94 response = state.get("response", {}) or {} 

95 return cls( 

96 state=state, 

97 response_type=response.get("response_type"), 

98 response_data=response.get("response_data"), 

99 response_message=response.get("response_message"), 

100 ) 

101 

102 @classmethod 

103 def error_result(cls, error: str, state: dict[str, Any] | None = None) -> "ExecutionResult": 

104 """Create an error ExecutionResult. 

105  

106 Args: 

107 error: Error message 

108 state: Optional partial state 

109  

110 Returns: 

111 ExecutionResult with success=False 

112 """ 

113 return cls( 

114 state=state or {}, 

115 success=False, 

116 error=error, 

117 ) 

118 

119 def to_response_dict(self) -> dict[str, Any]: 

120 """Convert to API response dictionary. 

121  

122 Returns: 

123 Dict suitable for API response 

124 """ 

125 if not self.success: 

126 return { 

127 "type": "error", 

128 "error": self.error, 

129 } 

130 return { 

131 "type": self.response_type or "unknown", 

132 **(self.response_data or {}), 

133 }