Coverage for src / mcp_server_langgraph / observability / langsmith.py: 100%

1""" 

2LangSmith configuration and integration for LangGraph agent observability 

3""" 

4 

5import os 

6from typing import Any 

7 

8from mcp_server_langgraph.core.config import settings 

9 

10 

11def configure_langsmith() -> bool: 

12 """ 

13 Configure LangSmith tracing using environment variables. 

14 

15 Returns: 

16 bool: True if LangSmith is configured and enabled, False otherwise 

17 """ 

18 if not settings.langsmith_tracing: 

19 return False 

20 

21 # Set LangSmith environment variables for automatic tracing 

22 if settings.langsmith_api_key: 

23 os.environ["LANGSMITH_API_KEY"] = settings.langsmith_api_key 

24 os.environ["LANGCHAIN_API_KEY"] = settings.langsmith_api_key # Alias 

25 

26 os.environ["LANGCHAIN_TRACING_V2"] = str(settings.langsmith_tracing_v2).lower() 

27 os.environ["LANGCHAIN_PROJECT"] = settings.langsmith_project 

28 os.environ["LANGCHAIN_ENDPOINT"] = settings.langsmith_endpoint 

29 

30 print("✓ LangSmith tracing configured") 

31 print(f" - Project: {settings.langsmith_project}") 

32 print(f" - Endpoint: {settings.langsmith_endpoint}") 

33 print(f" - Tracing V2: {settings.langsmith_tracing_v2}") 

34 

35 return True 

36 

37 

38def get_run_metadata( 

39 user_id: str | None = None, request_id: str | None = None, additional_metadata: dict[str, Any] | None = None 

40) -> dict[str, Any]: 

41 """ 

42 Create metadata for LangSmith runs. 

43 

44 Args: 

45 user_id: User identifier 

46 request_id: Request identifier for correlation 

47 additional_metadata: Additional custom metadata 

48 

49 Returns: 

50 Dictionary of metadata for LangSmith runs 

51 """ 

52 metadata = { 

53 "environment": settings.environment, 

54 "service_name": settings.service_name, 

55 "service_version": settings.service_version, 

56 "model_name": settings.model_name, 

57 "llm_provider": settings.llm_provider, 

58 } 

59 

60 if user_id: 

61 metadata["user_id"] = user_id 

62 

63 if request_id: 

64 metadata["request_id"] = request_id 

65 

66 if additional_metadata: 

67 metadata.update(additional_metadata) 

68 

69 return metadata 

70 

71 

72def get_run_tags(user_id: str | None = None, additional_tags: list[str] | None = None) -> list[str]: 

73 """ 

74 Create tags for LangSmith runs. 

75 

76 Args: 

77 user_id: User identifier to include as tag 

78 additional_tags: Additional custom tags 

79 

80 Returns: 

81 List of tags for the run 

82 """ 

83 tags = [ 

84 settings.environment, 

85 settings.llm_provider, 

86 f"model:{settings.model_name}", 

87 ] 

88 

89 if user_id: 

90 tags.append(f"user:{user_id}") 

91 

92 if additional_tags: 

93 tags.extend(additional_tags) 

94 

95 return tags 

96 

97 

98class LangSmithConfig: 

99 """LangSmith configuration helper class""" 

100 

101 def __init__(self) -> None: 

102 self.enabled = configure_langsmith() 

103 

104 def is_enabled(self) -> bool: 

105 """Check if LangSmith is enabled""" 

106 return self.enabled 

107 

108 def get_client_kwargs(self) -> dict[str, Any]: 

109 """ 

110 Get kwargs for LangSmith Client initialization. 

111 

112 Returns: 

113 Dictionary of client configuration 

114 """ 

115 if not self.enabled: 

116 return {} 

117 

118 return { 

119 "api_key": settings.langsmith_api_key, 

120 "api_url": settings.langsmith_endpoint, 

121 } 

122 

123 def create_run_config( 

124 self, 

125 run_name: str, 

126 user_id: str | None = None, 

127 request_id: str | None = None, 

128 tags: list[str] | None = None, 

129 metadata: dict[str, Any] | None = None, 

130 ) -> dict[str, Any]: 

131 """ 

132 Create a run configuration for LangChain/LangGraph execution. 

133 

134 Args: 

135 run_name: Name of the run 

136 user_id: User identifier 

137 request_id: Request identifier 

138 tags: Custom tags 

139 metadata: Custom metadata 

140 

141 Returns: 

142 Run configuration dictionary 

143 """ 

144 config = { 

145 "run_name": run_name, 

146 "project_name": settings.langsmith_project, 

147 "tags": get_run_tags(user_id, tags), 

148 "metadata": get_run_metadata(user_id, request_id, metadata), 

149 } 

150 

151 return config 

152 

153 

154# Global instance 

155langsmith_config = LangSmithConfig()